Stefan Brenner
email: brenner@ibr.cs.tu-bs.de

Tobias Hundt
email: hundt@ibr.cs.tu-bs.de

Giovanni Mazzeo
email: giovanni.mazzeo@uniparthenope.it

Rüdiger Kapitza
email: rrkapitz@ibr.cs.tu-bs.de

Secure Cloud Micro Services using Intel SGX

Keywords: Vert.x, SGX, Cloud Security, Micro services

The micro service paradigm targets the implementation of large and scalable systems while enabling fine-grained service-level maintainability. Due to their scalability, such architectures are frequently used in cloud environments, which are often subject to privacy and trust issues hindering the deployment of services dealing with sensitive data. In this paper we investigate the integration of trusted execution based on Intel Software Guard Extensions (SGX) into micro service applications. We present our Vert.x Vault, that supports SGX-based trusted execution in Eclipse Vert.x, a renowned tool-kit for writing reactive micro service applications. With our approach, secure micro services can run alongside regular ones, inter-connected via the Vert.x event bus to build large Vert.x applications that can contain multiple trusted components. Maintaining a full-fledged Java Virtual Machine (JVM) inside an SGX enclave is impractical due to its complexity, less secure because of a large Trusted Code Base (TCB), and would suffer from performance penalties due to a high memory footprint. However, as Vert.x is written in Java, for a lean TCB this requires integration of native enclave C/C ++ code into Vert.x, for which we propose the usage of Java Native Interface (JNI). Our Vert.x Vault provides the benefits of micro service architectures together with trusted execution to support privacy and data confidentiality for sensitive applications in the cloud at scale. In our evaluation we show the feasibility of our approach, buying a significantly increased level of security for a low performance overhead of only ≈ 8.7%.

Introduction

Micro services are popular as they offer a new paradigm and many benefits such as flexibility, scalability, ease of development and manageability of applications [START_REF]What Led Amazon to its Own Microservices Architecture[END_REF]. Due to their scaling nature and flexibility, micro service architectures are mostly used in data centre and cloud scenarios where scale out capabilities are required to handle high load. However, trust issues still hinder the widespread adoption of cloud services and the deployment of sensitive applications processing sensitive data in the cloud [START_REF] Jayaram | Trustworthy Geographically Fenced Hybrid Clouds[END_REF][START_REF] Pearson | Privacy, Security and Trust Issues Arising from Cloud Computing[END_REF].

With Software Guard Extensions (SGX) [START_REF] Anati | Innovative Technology for CPU Based Attestation and Sealing[END_REF][START_REF] Mckeen | Innovative Instructions and Software Model for Isolated Execution[END_REF], Intel recently released a new technology for protecting applications from many-even physical-attacks such as the cold boot attack. SGX is an instruction set extension, released in the Skylake processor family, that allows the creation of Trusted Execution Environment (TEEs) inside the address space of an application. SGX TEEs are called enclaves and provide strong protection of code and data inside through encryption and integrity checks of their memory range directly by the CPU. This allows a strong adversary model and to limit the trusted computing base to the enclave code and the CPU package only, which is especially useful for sensitive data processing in the cloud. On currently available hardware, SGX is limited to a maximum of 128MB of memory to be used for enclaves. This limitation requires to keep the memory footprint of enclaves as narrow as possible in order to prevent significant performance implications. Another important aspect with regards to security of TEEs is to keep their Trusted Code Base (TCB) as small as possible. This is due to the fact, that larger amounts of code usually lead to more exploitable security vulnerabilities [START_REF]Open Source Report[END_REF].

Amongst others like Spring Boot1 , Go Micro2 , an excellent example for a renowned protagonist in the context of micro service development is Eclipse Vert.x [2]. The Vert.x tool-kit introduces the notion of verticles and supports the development of micro service applications. Furthermore, it provides a distributed event bus for its verticles to communicate with each other in a reactive fashion.

In this paper, we investigate the protection of data confidentiality in micro service applications by exploitation of the SGX technology, and chose the Vert.x tool-kit as an example micro service environment. Our contribution comprises the design of our Vert.x Vault, that allows the integration of slim SGX TEEs into the JVM-based Vert.x tool-kit, and demonstrates the feasibility of our approach, as well as an evaluation of its induced performance overhead.

A technical challenge to be solved in this work was the way of integration of native C/C ++ -based TEEs into the Java/Java Virtual Machine (JVM)-based environment of the Vert.x tool-kit. As we argued in earlier work [START_REF] Brenner | SecureKeeper: Confidential ZooKeeper using Intel SGX[END_REF], porting a full-fledged JVM into a TEE is not only a complex endeavour, but also violates common security principles such as the size and complexity of the TCB as well as the memory footprint of the TEEs which is highly performance-critical. Moreover, in this work we elaborate why the usage of SGX-based TEEs is a good fit for micro service applications.

In Section 2 we describe SGX and Vert.x as the cornerstone of our system. Next, in Section 3 we show the design of our Vert.x Vault followed by implementation details in Section 4. Afterwards, in Section 5, we present the benefits of our Vert.x Vault adopted in a critical infrastructure use case scenario. Finally, we measure the performance of our approach in Section 6 and conclude in Section 8.

Background

In this section we describe the architecture, purpose, and main concepts of the Vert.x tool-kit. In addition, we give a short introduction to the SGX technology and describe the enclave life cycle and programming model.

V 1 V 2 V 3 V 4 V 5 T 1 Worker 1 T 2 Worker 2 T 3 EventLoop Host 1 V 1 V 2 V 3 V 4 V 5 T 1 Worker 1 T 2 Worker 2 T 3

EventLoop

Host 2

Event Bus

Fig. 1: Vert.x architecture and thread model (Verticles V 1-5 , Threads T 1-3 ).

Eclipse Vert.x

The rationale behind the concept of a micro service is to do one small thing and reduce complexity. In a micro service architecture, multiple of these services interact with each other and each of them has its very limited purpose. This allows the flexibility of development of such architectures, and also large development teams to collaborate simultaneously. Real applications of Vert.x can be for example REST services, or real time web applications 3 . According to its authors, "Vert.x is a tool-kit for building reactive applications on the JVM" [2]. In the context of Vert.x, micro services are called verticles, and are supposed to comprise a scarce, well-defined part of application logic. As Vert.x is a polyglot tool-kit for JVM-based languages, verticles can be implemented in various programming languages and interact across programming languages. Verticles communicate via the event bus that connects verticles even across machine boundaries were they can subscribe an "address" in order to receive callbacks once a message arrives for this address.

All verticles are scheduled via the event loop thread of Vert.x (one per physical CPU core), which delivers events to verticles. Supporting the idea of reactive applications, a verticle should never block this thread but implement blocking operations such as I/O operations in an asynchronous fashion. Long running tasks can be done in a worker verticle on a separate thread pool. Figure 1 illustrates the overall Vert.x system architecture and thread model.

Currently, there is no support for trusted execution in Vert.x, however, the Vert.x engineers recently released a secure event bus implementation that provides protection of exchanged event bus messages using Transport Layer Security (TLS).

SGX

SGX [START_REF] Anati | Innovative Technology for CPU Based Attestation and Sealing[END_REF][START_REF] Mckeen | Innovative Instructions and Software Model for Isolated Execution[END_REF] is a new instruction set extension by Intel, which has been released in the Skylake processor generation. It allows the creation of TEEs-called enclavesinside the address space of user space applications. Enclaves are protected by the CPU package itself from a number of critical security threats: enclave memory is transparently encrypted and integrity-checked by the CPU, limiting trust to only the CPU package itself. Entering and exiting an enclave is only possible via a defined enclave interface, that describes entry points and the allowed number of concurrent threads inside an enclave. Calls to enclave functionality are called enclave calls (ecalls), while calls from an enclave to outside code are called outside calls (ocalls). The enclave interface is described in a domain-specific language during development, which is also used to generate untrusted and trusted ecall and ocall stubs. Figure 2 illustrates the control flow of an ecall x, an ocall y and returning from an ecall to the untrusted code z.

Enclave memory is backed by a range of normal DRAM-called Enclave Page Cache (EPC)-which is reserved by the firmware during the boot process. This memory range is managed by the untrusted operating system, while its contents are encrypted by the CPU package with a random key. Currently, SGX supports a maximum of 128 MB of EPC for all enclaves running on a system together. Memory demand exceeding this range requires the SGX driver to reencrypt enclave pages and copy them to regular system memory causing a high performance impact.

Design

With our Vert.x Vault we want to integrate TEEs into Vert.x applications such that the TEE developer can offload parts of the application logic to SGX enclaves. This allows usage of SGX for critical parts of the application logic while keeping most of it including the code base of Vert.x untrusted and not part of the TCB. Vert.x and SGX TEEs are a suitable fit, as the micro service paradigm matches the original idea of the SGX engineers to keep enclaves lightweight and small, and the ability of SGX to integrate multiple TEEs into a single user process.

Since Vert.x is written in Java and enclaves are written in C/C ++ , we propose integration of enclaves via Java Native Interface (JNI) into verticles. While the integration of a full-fledged JVM into an enclave would be possible, it is not favourable as it introduces a lot of complexity to the enclave and causes a drastic increase of the TCB, leading to a higher probability of exploitable security vulnerabilities and an increased attack surface. Furthermore, in the context of SGX this would lead to a significant performance penalty as the available enclave memory-limited to 128 MB on current hardware-would be exceeded. Exceeding the enclave memory causes the SGX driver to move enclave pages to regular system memory, requiring an expensive re-encryption of the pages [START_REF] Brenner | SecureKeeper: Confidential ZooKeeper using Intel SGX[END_REF].

An overview of our architecture is shown in Figure 3, illustrating two verticles connected via the Vert.x event bus, and one of them containing an SGX enclave. As can be seen from the figure, a secure verticle is a verticle with an integrated enclave, that can be reached by other verticles via the Vert.x event bus. This enables the developer to design large micro service applications consisting of a mix of multiple untrusted and secure verticles. It is also possible to explicitly guide further isolation via process boundaries, as verticles can be deployed on specific hosts and multiple verticles can reside in the same JVM process.

The enclave which is integrated into a secure verticle, contains our Vert.x Vault as a C/C ++ library offering an Application Progamming Interface (API) to access the event bus directly from the enclave. In order to integrate the C/C ++based enclave into the Java-based Vert.x tool-kit, we use a small JNI component to forward the Vert.x Vault calls to the original Vert.x API. By this, the enclave is able to register an address on the event bus and receive messages from other verticles, as well as sending messages to arbitrary addresses on the event bus.

Adversary Model and Assumptions

By using SGX to implement TEEs we gain a very high level of security allowing a very strong adversary model. We assume an attacker that has full access to the machines running Vert.x, including the firmware, Operating System (OS), and all system and user space software. Even physical attacks such as cold boot attacks can be allowed without violating the security guarantees that SGX provides. We assume a correct implementation of SGX and all cryptographic primitives.

Security Aspects

As already outlined, the size of the TCB influences the overall security of an application. Besides its complexity and the fact that syscalls can not be done in an enclave, this is the main security reason why a full-blown JVM should not be ported to run inside an enclave. Running a JVM inside an enclave would lead to both, a large TCB inside the enclave as well as a large memory footprint of the enclaves which negatively affects both, security and performance.

Hence, in our approach we aim at C/C ++ -based enclaves that execute only minimal parts of the application logic in the trusted environment. Partitioning the application logic to decide which parts of application logic are required to run inside the enclave and which not is the task of the developer. We demonstrated this approach-called application partitioning approach-for ZooKeeper, a complex Java-based coordination service [START_REF] Brenner | SecureKeeper: Confidential ZooKeeper using Intel SGX[END_REF], in an earlier work.

Running parts of application logic in an enclave, provides protection of confidentiality and integrity of data inside the enclave and the enclave code itself. However, once data is exchanged between and enclave and the untrusted outside world, data must also be protected in an adequate way; traditionally by using using TLS between the two parties communicating. However, in case of SGX, the TLS endpoint must reside inside the enclave. In our Vert.x Vault, the enclave integrated into a secure verticle inherits the Intel SGX SDK features and can resort to the included cryptographic functions in order to protect data exchanged via the Vert.x event bus or with other secure verticles, respectively.

Enclave Interface Another possible security aspect is the enclave interface, i.e. the amount and signature of ecalls. In general, there will be less security vulnerabilities with a more narrow enclave interface. Also, we wanted to keep the enclave interface generic in order to support any kind of application logic running in the enclave. The complete enclave interface is illustrated in Listing 1.1.

We defined an enclave entry method which is called after the enclave is created. This allows the implementation of initialisation routines inside the enclave and to register to event bus addresses on secure verticle start-up and is analogous to the verticle's constructor method. Next, we implemented another method to deliver messages from the event bus to the enclave. Only within the enclave we distinguish the address on the event bus and deliver the message to the respective callback function. Calls from the enclave to the untrusted context are also part of the enclave interface. Firstly, there are two methods for registering and de-registering an address on the Vert.x event bus. During registration, the enclave stores a function pointer in an internal (trusted) data structure to the user-defined callback function once a message is received for this address. In order to allow sending messages to the event bus, we implemented two ocalls to send and broadcast messages onto the event bus respectively.

Programming Model

A crucial requirement of our project was to give the enclave developer the impression of the Vert.x programming model inside an enclave. Consequently, we defined the notion of a secure verticle that essentially represents a verticle whose application logic is implemented inside an enclave. As a Vert.x application will usually consist of multiple verticles, the idea is to offload all sensitive application logic components to secure verticles that can interact with the other verticles via the event bus. In terms of the application partitioning approach from our earlier work [START_REF] Brenner | SecureKeeper: Confidential ZooKeeper using Intel SGX[END_REF], only application logic fragments that require direct plain text access to user data should be implemented as secure verticles, whereas all other parts should be untrusted verticles. This is compliant with our goal of minimising the TCB for a higher level of security.

Inside enclaves of a secure verticle, we want to give the developer a Vert.x-like programming model, i.e. reactive event-based callbacks. For this purpose, we mimic the Vert.x event bus API inside the enclave, and allow the enclave developer to register event bus addresses and receive callbacks for the registered addresses as well as sending messages to the event bus. By this, the secure verticles get integrated into the complete system of verticles of a full Vert.x application.

The integration of SGX enclaves into Java applications in general, and into the Vert.x tool-kit specifically in this work, can be done using JNI. For this purpose we implement the interface that the enclave uses to interact with the event bus, as well as the one that Vert.x uses to forward incoming events to the enclave in JNI. In conjunction with stub code generated by the SGX Software Development Kit (SDK) supporting interaction of untrusted C/C ++ code and the enclave code, this requires additional copies of the message buffers. We discuss their performance impact in our evaluation in Section 6.

In order to support multiple secure verticles coexisting in the Vert.x environment, we need to maintain an enclave identifier in the Java class representing the secure verticle, such that each secure verticle "knows" its associated enclave.

Furthermore, we want to enable one secure verticle to register multiple individual callback functions inside the enclave for different event bus addresses. This requires to maintain a distinct entry point inside the enclave for all incoming messages where a lookup happens in order to find the right callback function for this event. Including the deregistration of addresses on the event bus, all these features are implemented as part of our Vert.x Vault.

Bootstrapping and Remote Attestation

SGX enclaves inherently are not able to contain secrets when they are created, thus injection of secrets into enclaves must be done after a successful Remote Attestation (RA). Integration of RA into our Vert.x Vault architecture requires a generic untrusted component inside of each secure verticle that can create a "quote" of the enclave and publish it on the event bus to an administrator component. Along with a public key generated by the enclave, the quote can be used by an administrator to verify the secure enclave remotely and via the Vert.x event bus by using the Intel attestation service [START_REF] Anati | Innovative Technology for CPU Based Attestation and Sealing[END_REF]. After a successful RA, the administrator can inject secrets into the enclaves of secure verticles.

Secure Event Bus

Recently, the Vert.x developers added the secure event bus to Vert.x, that allows the encryption of all messages on the event bus transparently to the verticles. Essentially, the secure event bus provides TLS-based transport security of messages across machine boundaries. In order to be transparent to verticles, the integration of the secure event bus is done deeply inside Vert.x. Hence, the verticles can not notice the encryption or-in case of a secure verticle using our Vert.x Vault-move the encryption endpoints into the enclaves. For this reason, we consider the secure event bus orthogonal to the implementation of our Vert.x Vault. In order to integrate it with our secure verticles, the TLS endpoint of the secure event bus must be moved inside the enclave, requiring major changes to the internals of Vert.x. However, even without integration of the secure event bus with secure verticles, the latter can of course establish their own secure connections transparently to the event bus.

In the context of the SERECA project, we demonstrate the opportunities coming from an integration of SGX with Vert.x through a suitable use case scenario: a cloud-based application for the monitoring of seven dams belonging to a water supply network [START_REF] Cerullo | A secure cloudbased SCADA application: The use case of a water supply network[END_REF]. This use case unarguably belongs to the Critical Infrastructure (CI) domain. CIs enclose assets essential for the functioning of all countries' fundamental facilities such as energy, telecommunications, water supply, and transport. Due to their importance in nations' sustainability, CIs are target of terrorist cyber-attacks, as demonstrated by recent events: "Black Energy"4 in 2015, and "Havex"5 in 2014. Security threats related to the disclosure or manipulation of sensitive data slows the migration of CIs to cloud technologies.

Through the cloud-based Water Supply Network Monitoring (WSNM) application, we want to prove that the combination of Vert.x with SGX-enabled by our Vert.x Vault-can represent a possibility to overcome CI's security concerns. The different tools provided by Vert.x enable the development of a reactive WSNM application based on micro-services having the following peculiarities: 1) Easily deployable among the dams and the cloud; 2) Highly scalable in front of sensors measurements peaks 3) Highly available in front of failures; 4) Highly performant in the process of sensors data collection, elaboration and provision. Moreover, the SGX extension allows the encryption/decryption and sealing of sensitive measurements by an enclave using a platform-specific key and while requiring to trust only the CPU package.

Figure 4 shows the overall architecture of the WSNM distributed application. On the dam-side, a data collector verticle interfaces, through a ModBus protocol, with a data logger equipment responsible for providing all the sensor data. Then, the acquired measurements are sent to the registered cloud-verticles through the Vert.x secure event bus. Thanks to this, messages are securely encrypted and signed. Only the involved receiver verticles, knowing the encryption key (AES) and integrity key (HMAC), can decrypt the message. In addition to a TLS secure communication, a Route-based encryption is enforced. Senders and receivers do not indicate the key in the message. They use an equivalent configuration, which defines the key to be used for a specific address. In this way, they do not share key names, but agree on the address-key mapping.

On the cloud-side, four registered verticles receive data through the secure event bus and, based on their duties, take a specific action on it. Two of them are secure verticles and so make use of Vert.x Vault APIs seen in Section 3.3:

Cache Archiver Verticle (CAV): It is responsible for the storage of time-recent data into an in-memory system. Such a data is needed, e.g., by the alarm manager in charge of real-time analysis activities. The CAV is unarguably a Secure Verticle as the measurements it stores in memory must be encrypted. Therefore, the CAV registers itself from within the enclave to the Vert.x secure event bus through a ocall register. Then, when new updates are received, the data-before being Alarm Manager Verticle (AMV): The alarm manager is in charge of signalling dangerous situations occurring on the dam infrastructure by enforcing a Complex Event Processing (CEP), i.e., correlating different data sources to find events or patterns suggesting more complicated circumstances. To do that, the AMV needs to receive live data from the collector verticle and also temporarily store data in memory. For this reason, it communicates with the CAV through point-to-point messages exchanged through the secure event bus. The AMV, that currently operates on data securely stored, will be extended as a secure verticle to realize the CEP processing into an enclave.

Storage Archiver Verticle (SAV): Beside saving data for CEP processing purposes, it is also important to store measurements for historical trend evaluations into a persistent storage system. Even in this case it is important that stored data is encrypted. Hence, even the SAV is defined as a Secure Verticle.

Web Proxy Verticle (WPV): The final user operator has access to the monitoring application through a web-based dashboard, which reports real-time measurements, historical trends, and alarm notifications. Such a dashboard communicates with the back-end cloud application through the WPV, which can ask any verticle for data to be sent to the web browser based on the requests. The TLS-enabled transmission of sensitive data is realized through a Vert.x secured SockJS bridge-provided in the secure event bus-able to encrypt/decrypt data at application level using a key shared with the web browser.

Evaluation

In order to measure the performance impact of enclaves integrated into verticles using our Vert.x Vault, we wrote a benchmark tool in order to evaluate the performance of our prototype and present the results in this section. All measurements presented in this section were done with real enclaves and executed on SGX-capable machines with Core i7-6700 @3.4GHz, 24 GB RAM, 256 GB SSD and 4x GbE.

Performance Measurement

The measurement scenario comprises three verticles deployed on two hosts: a sender verticle that sends requests and measures the response time and throughput on one host, and a regular and a secure verticle both running on the other host answering the requests. All communication between verticles is transferred via the Vert.x event bus using the Hazelcast cluster manager6 , which is the default cluster manager implementation of Vert.x. In order to get an impression of the performance impact of the usage of SGX in verticles, we exchange payloads of various size between the verticles and measure throughput and response time.

In a first experiment we investigate the level of concurrency that leads to optimal throughput by increasing the maximum allowed number of concurrent pending requests on the event bus in the sender verticle. We chose a realistic payload size of real Vert.x applications of 1024 Bytes for this experiment. The results of this experiment are shown in Figure 5 which also includes the response times. As can be seen, the value of 128 pending requests leads to the highest throughput, while higher values do not significantly increase throughput.

We also measure the throughput of regular and secure verticles for various sizes of message payload using the aforementioned value of 128 concurrently pending requests that led to optimal throughput. Figure 6 illustrates the request throughput of our experiment for various payload sizes. It proves that secure verticles perform quite well, and in most of the cases reach a throughput almost as high as regular verticles. The same is true for the transmitted payload traffic on the event bus between the verticles, as can be seen from the same Figure 6.

Finally, we measured the mean response time of requests for regular and secure verticles as illustrated in Figure 7. The figure shows the additional processing time required to enter the enclave and copy the payload buffer in and out. As expected, in general the response time of secure verticles is higher than the one of regular verticles. For very small payloads between 0 -512 Bytes there is a relatively high overhead of ≈ 83.1% of secure verticles, while for larger payloads between 512 Bytes and 256 KBytes the relative overhead decreases to only ≈ 26.2%. We explain this by the constant overhead of entering and exiting the enclave which is relatively more notable for small payloads.

Size of TCB

As motivated earlier in this paper, we aimed at minimising the Source Line of Code (SLOC) inside the enclave (i.e. the TCB) in order to optimise both performance and security. While the Vert.x tool-kit's code base comprises thousands of SLOC, our Vert.x Vault only requires 69 lines of code inside the enclave in order to enable the enclave code to access the Vert.x event bus. Apart from this, the TCB naturally comprises the code by the micro service application developer, but the majority of code resides outside of the enclave and untrusted. This enables the developer to decide on a very fine-grained level what code is trusted.

Related Work

Haven [START_REF] Baumann | Shielding Applications from an Untrusted Cloud with Haven[END_REF] was the first system and supports execution of legacy applications unchanged in an SGX enclave. While running legacy applications in enclaves has many benefits, such as support for closed source applications, this feature adds a large amount of code to the TCB of the enclave, like a full library OS and other libraries. This eventually leads to a large memory footprint, which negatively influences the performance of the application on top of SGX.

SCONE [START_REF] Arnautov | SCONE: Secure linux containers with Intel SGX[END_REF] supports running secure containers based on Docker [1]. This approach reduces the memory footprint of the enclaves when compared to Haven, while still featuring a generic platform for the execution of secure containers.

Another type of related work is the field of partitioned applications, that are specifically paritioned applications to run most efficiently and with the least possible memory footprint on top of SGX. An example for securing complex cloud services by partitioning them for usage with SGX enclaves is SecureKeeper [START_REF] Brenner | SecureKeeper: Confidential ZooKeeper using Intel SGX[END_REF][START_REF] Brenner | Running ZooKeeper Coordination Services in Untrusted Clouds[END_REF] featuring a partitioned Apache ZooKeeper [START_REF] Hunt | ZooKeeper: Wait-Free Coordination for Internet-Scale Systems[END_REF] coordination service.

The communication of verticles on the Vert.x event bus is alike publishsubscribe systems. In this context, Pires et al. presented secure content-based routing with SGX [START_REF] Pires | Secure Content-Based Routing Using Intel Software Guard Extensions[END_REF]. Their system protects the subscription process of clients at data providers and the confidentiality of the data exchanged between them. In contrast, our Vert.x Vault not only enables protecting the confidentiality of exchanged messages between verticles, but also allows general-purpose sensitive data processing of inside secure verticles.

Conclusion

Micro service architectures are a modern way of writing scalable applications tailored for cloud environments. However, trust issues still hinder the adoption of cloud technology especially for sensitive applications handling sensitive data.

In this work, we showed an approach of integrating TEEs into the micro service tool-kit Vert.x. With our approach due to the usage of SGX, we gain a high level of security at a strong adversary model. Furthermore, we support the Vert.x programming model and control flow inside the TEE by mimicing its API. This allows interaction of trusted code with the Vert.x event bus.

Our prototype implementation demonstrates the feasibility of our approach, while the evaluation shows a very low performance overhead of only ≈ 8.7%.

Fig. 2 :

 2 Fig. 2: Interaction between untrusted application and SGX enclave.

Fig. 3 :

 3 Fig. 3: Architectural Overview of Vert.x Vault showing two verticles connected via the Vert.x event bus and one of them containing an SGX enclave.

  void e c a l l _ e n c l a v e I n i t () ; public void e c a l l _ d e l i v e r M s g ( [ in , size = lenCh ] const char * channel , size_t lenCh , [ in , size = lenMsg ] const char * msg , size_t lenMsg ) ; }; untrusted { void ocall_r egister ( [ user_check ] void * weak , [ in , string ] const char * channel , size_t len ) ; void o c a l l _ u n r e g i s t e r ( [ user_check ] void * weak , [ in , string ] const char * channel , size_t len ) ; void ocall_send ( [ user_check ] void * weak , [ in , string ] const char * channel , size_t len , [ in , string ] const char * msg , size_t lenMsg ) ; void o ca ll _b r oa dc a st ( [ user_check ] void * weak , [ in , string ] const char * channel , size_t len , [ in , string ] const char * msg , size_t lenMsg ) ; }; }; Listing 1.1: Enclave interface description.

Fig. 4 :

 4 Fig. 4: Architecture of the dam monitoring application SGX-enabled

Fig. 5 :

 5 Fig. 5: Request throughput for various number of pending requests.

Fig. 7 :

 7 Fig. 7: Response times for various payload sizes.

https://projects.spring.io/spring-boot/

https://github.com/micro/go-micro

https://ics-cert.us-cert.gov/alerts/ICS-ALERT-14-281-01B

https://ics-cert.us-cert.gov/alerts/ICS-ALERT-14-176-02A

https://hazelcast.com/

Acknowledgements

This project received funding from the European Union's Horizon 2020 research and innovation programme under the SERECA project (Grand No. 645011).


Fábio Coelho
email: fabio.a.coelho@inesctec.ptjop

Miguel Matos

José Pereira

Rui Oliveira

Similarity Aware Shuffling for the Distributed Execution of SQL Window Functions

Window functions are extremely useful and have become increasingly popular, allowing ranking, cumulative sums and other analytic aggregations to be computed over a highly flexible and configurable sliding window. This powerful expressiveness comes naturally at the expense of heavy computational requirements which, so far, have been addressed through optimizations around centralized approaches by works both from the industry and academia. Distribution and parallelization has the potential to improve performance, but introduces several challenges associated with data distribution that may harm data locality. In this paper, we show how data similarity can be employed across partitions during the distributed execution of these operators to improve data co-locality between instances of a Distributed Query Engine and the associated data storage nodes. Our contribution can attain network gains in the average of 3 times and it is expected to scale as the number of instances increase. In the scenario with 8 nodes, we were to able attain bandwidth and time savings of 7.3 times and 2.61 times respectively.

Introduction

Nowadays, the scalability of database engines is paramount, specifically when it is targeted at large scale analytical processing. Systems must be able to support several computing nodes, enabling component scalability to possibly reach hundreds or thousands of nodes. However, reaching such scale introduces several challenges associated with data and request distribution and balance. Cloud computing infrastructures offer a nearly transparent environment where computation is available as virtually infinite computing nodes. However, commercial relational database engines (RDBMS) do not conform to such paradigm, typically offering a monolithic structure. Legacy-type servers are usually considered for running RDBMSs, limiting system scalability from the purchase moment or until they become economically unacceptable.

Window Functions (WF) define a sub-set of analytical operations that enable the formulation of analytical queries over a derived view of a given relation R. They are also known as OLAP Analytical Functions and are part of the SQL:2003 standard. All major database systems like Oracle [START_REF] Corporation | SQL analysis and reporting[END_REF], IBM DB2 [START_REF]OLAP specification[END_REF], Microsoft SQL Server [START_REF] Coporation | Transact-SQL[END_REF], SAP Hana [START_REF]SAP: SAP HANA SQL reference[END_REF], Cloudera Impala [START_REF] Kornacker | Impala: A modern, open-source SQL engine for hadoop[END_REF] or Postgresql [START_REF]Postgresql: Advanced features -window functions[END_REF] have the ability to execute a sub-set of the available WFs.

WF are widely used by analysts as they offer a highly configurable environment together with a straightforward syntax. In fact, SQL WF are used in at least 10% of the queries in TPC-DS [START_REF]Transaction Processing Performance Council[END_REF] benchmark, a benchmark suite aimed to evaluate data warehouse systems. Despite their relevance, parallel implementations and optimizations considering this operator are almost non existing in the literature. While [START_REF] Cao | Optimization of analytic window functions[END_REF][START_REF] Zuzarte | Winmagic: Subquery elimination using window aggregation[END_REF][START_REF] Leis | Efficient processing of window functions in analytical SQL queries[END_REF] are notable exceptions, these works are targeted at many-core CPU centralized architectures that are substantially different from distributed architectures.

The nature of current centralized architectures do not typically take into account data distribution. This eases their processing models, but prevents them to scale beyond the limitations of the hardware that hosts them. The massively parallel nature that distribution approaches enable requires, however, to carefully address data distribution.

Having the right grasp on data placement allows to improve data movement, but requires additional mechanisms to maximize network efficiency.

In this paper we focus on WF, particularly exploring opportunities for their distributed execution. We propose a technique that exploits similarity between partitions as a metric that can be used to judiciously improve the affinity of data and computing nodes, consequently minimizing the data movement between computing nodes. Contributions: First, we demonstrate that it is possible to improve data forwarding by using partition similarity to chose the forwarding mechanism between Distributed Query Engine (DQE) workers. Second we present an experimental evaluation that confirms the merit of our approach. Roadmap: The remainder of this paper is organized as follows: Section 2 introduces WF. Section 3 introduces Distributed WF, describing their query execution plans and cost models. Section 4 presents our similarity technique, improving affinity between data and computing nodes. Section 5 evaluates our proposal. Section 6 presents related work and Section 7 concludes our work.

Window Functions

WF started to be largely adopted by database vendors from the 2011 revision of the SQL standard. These are powerful analytical operators that enable complex calculations such as moving, cumulative or ranking aggregations to be computed over data. WF are expressed in SQL semantics by the keyword OVER as shown in Figure 1. In the next Sections we will analyze each part of the query. Like other analytical operators, WF are required to reflect several concepts, namely: the processing order, the partitioning of results or the notion of the current row being computed. These design constraints are clearly translated from the syntax as seen in the previous example, and configure two main considerations as foundation for the WF environment. Firstly, WF are computed after most of the remaining clauses in the query (e.g., such as JOIN, WHERE, GROUP BY or HAVING), but immediately before any required final ordering (e.g., ORDER BY). Secondly, the analytical operator to be computed with the WF environment will create an output attribute that reflects, but does not modify or filter the input data present in the source relation. Therefore, the result-set will present the same cardinality of rows as in the source relation, but will have an additional attribute mapping the result.

Partitioning, Ordering and Framing

The WF environment can be decomposed into three stages, as depicted in Figure 2, defining the processing order: the partitioning (1), ordering (2) and framing (3) stages. Each stage is defined by specific clauses namely: the PARTITION BY and the ORDER BY that respectively create logical partitions of distinct data elements and afterwards develop an intra-partition sorting. The logical partitions are regulated by the mandatory argument of the PARTITION BY clause, defining the column attribute or expression that controls the partitioning. The partition clause resembles the behavior of the GROUP BY clause, but does not collapse all group members into a single row. The intra-partition ordering follows the partitioning stage and is also regulated by the mandatory column attribute or expression considered as argument for the ORDER BY clause. The ordering stage is very important for a set of non-cumulative analytical functions, that are the focus of our contribution, but also as it is the costliest operation in the environment [START_REF] Cao | Optimization of analytic window functions[END_REF].

Finally, the framing stage builds on the provided ordering, taking into account the current row being considered to introduce the concept of window or frame. The frame is built from a group of adjacent rows surrounding the current row and changes as the current row moves towards the end of the partition. The framing is set by either the ROWS BETWEEN or the RANGE BETWEEN clauses. The former considers n rows before and after the current row, while the latter restricts the window by creating a range of admissible values and, the current row is considered if the stored values fit in the provided range 3 .

The WF environment allows to combine different clauses, enabling the inclusion or exclusion of each clause type. For instance, it is possible to declare a WF with just a partitioning or ordering clause. If no partitioning clause is declared, the entire relation is considered as a single partition. If no ordering clause is declared, then the natural ordering of the relations key, or partitioning clause (if present) is considered. Moreover, each available analytical function may or not change the computation logic. Due to space constraints we do not characterize all the possible configurations of the WF environment. The interested reader should consult the 2003 and 2011 revisions of the ANSI SQL standard for further information [START_REF]ANSI: Information technology -database languages -SQL multimedia and application packages[END_REF].

Cumulative and Ranking Analytical Functions

The analytical set of functions currently available in most Query Engines (QE) can be classified into Cumulative or Ranking. Cumulative analytical functions or aggregates, are a group of functions that are not order-bound. That is, when they are computed within a WF, an ORDER BY clause is not required. The sum(x), avg(x) or count() are just some examples of this category of functions. Figure 3(a) depicts the result of computing a WF structured as "select analytical function() OVER (PARTITION BY A ORDER BY D) FROM table", but immediately before applying the requiring analytical function to a given relation. Figure 3(b) depicts the result of computing the previous WF with the sum(D) function. The result of a cumulative function is the same for all the members belonging in the same partition.

Ranking analytical functions, on the other hand, are order-bound. That is, the function requires the data to be ordered according to some criteria in order to output a deterministic result, and thus, the ordering clause is always required. The rank(), dense rank() or ntile() are just some examples of this category of functions. Figure 3(c) outputs the result of computing the previous WF with the rank() function, outputting a different result for each row in the partition.

The ordering requirement for the latter category of functions implies data co-locality in order to minimize the number of sorting steps needed to achieve intra-partition ordering [START_REF] Cao | Optimization of analytic window functions[END_REF]. In the remainder of this paper, we consider a WF computing a ranking analytical function with a single partition and ordering clause and no framing clause, since the rank function implicitly defines framing constraints.

RDBMS are built from several components, namely the QE and the Query Optimizer (QO). The former translates SQL syntax into a set of single operators. The latter considers several statistical techniques to improve the query execution plan of a query. In a nutshell, QEs split the execution of a query in two separate stages: the query planning and the query execution. During the first stage, the QE decides how the query is executed during the second stage, and which operators are used in such a query plan. This builds a complex multi-optimization problem that has to be executed in polynomial time.

The QO uses hints about data in the form of statistical approximations, allowing the query engine to optimize query execution based on the approximation cost of each individual operator in a given data set. When scaling from a single QE to a DQE, data partitioning techniques are necessary in order to distribute data among instances. The number of available computing nodes configures the installed Degree of Parallelism (DOP). However, non-cumulative analytical algorithms are order-bound, thus requiring that logical data partitions are co-located (i.e., they should live in the same storage node). If elements of a given logical partition are spread in a group of nodes it becomes impossible to sort each logical partition in just one step. The sorting in each data partition would induce a partial sorting that is not deterministic and that would prevent inter-partition parallelism. The QOs therefore need to adapt their cost models to reflect the data movement required in order to ensure co-locality of partitions during execution time of the operator.

Distributed Query Engines

The DQE takes advantage of data distribution in order to scale query execution. The present architecture is provided by a Highly Scalable Transactional PaaS [START_REF] Jimenez-Peris | Cumulonimbo: A highly-scalable transaction processing platform as a service[END_REF]. Each node in the system is split in two layers, the DQE itself and the storage layer, holding the data partitions to be manipulated by a given DQE instance. Particularly, the considered DQE is based on the Apache derby project [START_REF]Apache: The apache derby project[END_REF] and the storage layer is provided by Apache HBase [START_REF] George | HBase: The Definitive Guide: Random Access to Your Planet-Size Data[END_REF], working over the Hadoop Distributed File Systems (HDFS) [13].

The DQE instances are able to accept client query requests through a JDBC connection and generate the distributed query plan. This plan is then shared with all participating DQE instances. The data distribution in each Storage Node is typically accomplished by means of an Hash function, considering a single or a collection of attributes as key. The distribution of keys lies within the inner characteristics of the considered hash function, usually producing uniform distributions that evenly place tuples across all available storage nodes. Poorly chosen hash functions may result in data skew and should be tailored to each specific workload providing adjusted table splitting [START_REF] Cruz | Workload-aware table splitting for nosql[END_REF].

Figure 4 presents the simplified distributed query plan for a ranking analytical function. The following stage numbers resemble the ones depicted in Figure 4. With data partitioned in several nodes, each one will scan (1) its local partition. The partial results found in each node derive from the data partitioning required to distributed data. Data movement is then required in order to ensure that each logical partition created by the partitioning clause will reside in a single node for computation. This is achieved by the

Data Shuffling

Data movement during the execution of a WF query is required, ensuring that all the elements of each logical partition are in the same location. In order to judiciously for-ward data while minimizing at the same time the transfer cost, in [START_REF] Coelho | Holistic shuffler for the parallel processing of SQL window functions[END_REF] we introduced a mechanism that works together with the data transfer mechanism, a shuffler, promoting co-location of logical partitions. This is achieved by considering an histogram, characterizing the universe of elements present in each partition. Briefly, the histogram should hold the cardinality of each different element in each different column qualifier. The histograms referring to each node are then combined into a global histogram. The introduction of this mechanism along with the shuffler, allowed to forward data to the specific node that should process a given partition. Consider Figure 5 where a table similar to the table in Figure 3(a) was split in two partitions on the storage layer. This initial partitioning is defined by hashing the value of the nodes ol w qualifier and performing the arithmetic modulo between the hash result and the number of computing instances (Hash(value in ol w) % #Nodes). Guided by the query in Figure 1, the results were then ordered according to the qualifier ol d (the partitioning clause). Both nodes of the storage layer hold elements from the available three partitions in ol d (p1, p2, p3). According to the previously introduced, ol d partitions (p2) and (p3) in instance DQE w1 will be relocated to instance DQE w2 and, ol d partition (p1) will be relocated from instance DQE w2 to instance DQE w1.

(p1) (p2) (p3) (p1) (p2) (p3)
On the one hand, hash forwarding a single row at a time prevents batching several rows in a single request. On the other hand, due to the asynchronous nature of DQEs, latency is usually not the bottleneck and thus, data movement can be delayed until network usage can be maximized [START_REF] Gonc ¸alves | An RDMA middleware for asynchronous multi-stage shuffling in analytical processing[END_REF]. This enables the use of batching in order to improve network usage. A batch payload is formed by grouping rows that need to be forwarded to a common destination and it is regulated by a buffer within the shuffling mechanism, whose size and delivery timeout are configurable. Nevertheless, the use of this mechanism can prove to be a misfit in cases where workloads do not benefit from grouping data (i.e., logical partitions with reduced number of rows). Therefore, not having to delay data transmission reduces execution time. To understand up to what level a given logical partition may or not benefit from batching, we considered a correlation mechanism to guide such decision, identifying the logical partitions that are good candidates for forwarding data in batch.

QOs found in modern QEs use several statistical mechanisms to explore data features in order to improve query execution performance. Without them, independence assumptions between attributes are preserved, which commonly leads to under or over provisioned query plans, which is particularly undesirable in DQEs. As in real-world data, correlations between relation attributes are the rule and not the exception, the array of correlation or other algebraic extraction mechanisms in the literature is vast, namely [START_REF] Brown | BHUNT: Automatic discovery of fuzzy algebraic constraints in relational data[END_REF][START_REF] Fan | Conditional functional dependencies for capturing data inconsistencies[END_REF][START_REF] Liu | Exploiting soft and hard correlations in big data query optimization[END_REF]. Correlations can also be used in DQEs to improve how data distribution is handled. When logical data partitions need to be relocated in order to improve co-locality, the correlation between qualifiers in different locations of the storage layer can be explored to minimize the required data movement.

Algorithm 1 Similarity Aware Shuffling Mechanism In this paper, we introduce a similarity measure to quantify to what level the partitions of a given attribute held by different storage nodes are alike. Data partitions with high similarity are good candidates to be shuffled within a batch payload. This is so as a high similarity implies a high common number of partitions. On the other hand, data partitions with low similarity are better candidates to be immediately shuffled for their destination. This is so as they share a low number of common partitions. This is efficiently achieved through Algorithm 1. The similarity measure quantifies in a universe between 0 (not similar) and 1 (similar) how similar two attributes are, by considering the number of unique values in each attribute to compute the metric. The data required to compute this metric is already provided by the histogram introduced in previous work [START_REF] Coelho | Holistic shuffler for the parallel processing of SQL window functions[END_REF], bypassing the need to collect additional statistical data. This structure is characterized by a small memory footprint (few KB) and the update period dictated by the DQE administrator. This algorithm will be considered during the first shuffling stage (stage (2) of Figure 5). It will consider each logical partition (P(r)), the previously introduced Histogram (H) and a configurable similarity threshold. Three auxiliary procedures are considered. The SIMILARITY procedure computes the similarity measure from the set of unique values in the qualifiers considered as arguments. The BATCHSHUFFLING procedure marshals all the rows of partition P(r) and sends it to the destination worker dest. The HASHSHUFFLING procedure marshals a single row r i and sends it to destination dest.

1: P(r) =< r 0 , r 1 , r 2 ,
When the shuffler action is required, it consults the Histogram H to verify what is the optimal destination (DQE instance) from row r i . When the destination is a remote instance (line 15), the shuffling mechanism computes the similarity measure between the local (attr A ) and destination (attr B ) qualifiers (line 16). The partition P(r) is marshaled to the appointed destination when the observed similarity is above threshold t (line 18) (BATCHSHUFFLING), or each row r i is otherwise sent to destination (line 20) (HASHSHUFFLING). The parameter t sets a threshold above which rows are forwarded in batch to the destination instance. This parameter defaults to 0.5 meaning that if not modified, rows are batch forwarded if the origin contains at least half the number of unique partition values of the destination.

Evaluation

We validated that by batch shuffling tuples between DQE instances we would save bandwidth, improving execution time of the shuffling stage. We considered a synthetic data set and shuffled rows between distinct DQE instances. The data set used was extracted from the TPC-DS [START_REF]Transaction Processing Performance Council[END_REF], a benchmark suite tailored for data analytics. We extracted a single relation (web sales) which is composed of 35 distinct attributes, configuring TPC-DS with a scale factor of 50GB. This resulted in a relation with 9.4GB corresponding to 36 million rows.

The outcome of the mechanism we propose is directly related with the data distribution considered. In order to bound the outcome of our contribution in terms of the lower and upper performance bounds, we statistically analyzed the considered relation. The lower bound is set by not using the similarity mechanism. The upper bound is set by considering the relation attributes that would favor data distribution. This was achieved by identifying the placement key attribute, but also a candidate attribute to be the partitioning clause or shuffling key (PBK) of the WF. The placement key attribute will define the data distribution in each DQE Storage Node through the use of an Hash function, and the PBK will define the runtime partitioning within the WF environment.

The results are depicted in Figure 6. The top plot presents the number of partitions in each single attribute in the considered relation. That is, the number of unique values in each attribute. The bottom plot depicts the average cardinality of each partition. That is, the average number of elements in each group of unique values in each attribute. The ideal candidate attribute to become the relation placement key is the attribute that displays the highest partition number and at the same time holds the smallest cardinality, ensuring an even data distribution and reduced data skew. Observing both plots leads us to consider attribute with index 17 (ws order number), displaying the highest number of partitions, each one with a single element. On the other hand, the candidate attributes to be selected as WF PBK are the attributes that would hold at the same time a high number of partitions and high partition cardinality. These are good PBK candidates since they will induce a number of logical partitions that is above the configured DOP. The observation of the plots leads to identify as candidates the attribute indexes depicted in dark gray, from which we select attribute 0 (ws sold date sk) as PBK.

After the election for the PK and PBK keys, we conducted a second experiment to verify the computed similarity measure. Figure 7 depicts the results of applying the metric in two scenarios. In both cases, we consider our scenario to be built from several DQE instances and corresponding Storage Nodes. On all experiments, we considered only the communication layer of the DQE where our contribution is, thus avoiding the SQL parsing and optimization stages. Each data partition was computed by applying an Hash function with the elected PK dividing the data into as many partitions as configured DQE instances. We first considered the configuration with 2 instances A and B. In the experiment in the top plot we computed the similarity measure between the the PBK of location A and each distinct attribute in location B. It is possible to observe that attribute 0 in location B presents the highest similarity, followed by attribute two. These are also the only attributes that are above the set up threshold of 0.5 denoted by the horizontal line. The remaining attributes have a residual similarity measure. The bottom plot depicts a different configuration where attribute 15 was randomly chosen among all non candidate attributes. The similarity measure in this attribute is lower than our threshold, even though it seems to be equal given the logarithmic scale required to observe the remainder attributes. Therefore, the results achieved during the first configuration would induce the shuffler to use batching mechanisms to forward partitions among DQE instances, instead of hash forwarding. The latter would culminate in sending a single row at a time. In order to verify the impact of our contribution regarding network usage, we conducted an experiment to assess the magnitude of the network savings promoted. Namely, we considered configurations with 2, 4 and 8 DQE and Storage instances. The computing nodes were only set up with the communication layer responsible for the shuffling in the WF environment. Each node is comprised of commodity hardware, with an Intel i3-2100-3.1GHz 64 bit CPU with 2 physical cores (4 virtual), 8GB of RAM memory and one SATA II (3.0Gbit/s) hard drive, running Ubuntu 12.04 LTS as the operating system and interconnected by a switched Gigabit Ethernet network. During execution, each computing node acts as a DQE instance shuffler, forwarding data to the remainder instances. In a distributed deployment, the DQE instance will be co-located with other services (e.g., storage node) which will typically restrict the available memory to the DQE instance.

We evaluated two configurations where the first represents a baseline comparison, forwarding all data by hash shuffling, and a second where data is forwarded according to our similarity mechanism. The results depicted in Figure 8 are twofold. The similarity measure registered both a decrease in bandwidth and it also promoted a shorter execution period for the shuffling technique. This is the result of pairing the batch shuffling mechanism together with the proposed similarity measure. The savings induced come at a residual cost, since the statistical information is not collected for the single purpose of this improvement, nor it has to be updated in each query execution. The similarity measure technique only proved effective from the configuration with 4 instances onward, since it is only from that configuration that both bandwidth and execution time are lower than the baseline. For the configuration with only two nodes, the baseline technique proved to be better by both shortening the shuffling time and registered bandwidth. However, in the configurations with 4 and 8 nodes, the similarity measure was able to reduce the bandwidth and execution time when compared with the baseline approach. As the number of partitions in the system increase, each single partition becomes responsible for a shorter set of data, promoting bandwidth savings up to 7.30 times for the 8 node configuration. The previous experiment evaluated the shuffling mechanism by considering an attribute with ideal similarity measure and partitioning on the storage layer. In order to demonstrate the impact of selecting an attribute that do not favor an uniform distribution of data among data partitions, we conducted a second experiment that considered an attribute with poor partitioning properties (i.e., reduced number of partitions). The results consider the same component configuration, but selected attribute 15 (ws warehouse sk) for the partitioning. When selecting an attribute that lacks the desirable distribution, the logical partitions will present an imbalance, thus promoting a low similarity measure. Therefore, the shuffling mechanism will not be able to maximize network usage and will end up having to consider the HASHSHUFFLING mechanism to forward data. The results are not thoroughly presented due to space constraints. However, we point out that they are in line with the considered baseline results presented in Table 1, registering a bandwidth variance of ±4%. Moreover, even though we do not consider it, the use of compression techniques may further increase the observed savings.

Related Work

Window Functions were introduced in the 2003 SQL standard. Despite its relevance, parallel implementations and optimizations considering this operator are almost non existing. Works such as [START_REF] Cao | Optimization of analytic window functions[END_REF] or [START_REF] Zuzarte | Winmagic: Subquery elimination using window aggregation[END_REF] fit in the first category, respectively tackling optimization challenges related with having multiple window functions in the same query, and showing that it is possible to use them as a way to avoid sub-queries and lowering quadratic complexity. However, such approaches do not offer parallel implementations of this operator. A vast array of correlation mechanisms have been so far deeply studied in the literature. Nonetheless, most of the conducted studies focus on efficient ways to discover and exploit soft and hard correlations [START_REF] Ilyas | CORDS: automatic discovery of correlations and soft functional dependencies[END_REF], allowing to find different types of functional dependencies. Works like [START_REF] Leis | Efficient processing of window functions in analytical SQL queries[END_REF] introduced mechanisms to improve the performance of the WF environment when many-core architectures are used. Distinct approaches and algorithm improvements are introduced, enabling to parallelize the distinct stages of the operator.

When addressing WFs, a common misconception generally brings a comparison between SQL WF (in which our contribution focuses) and CEP windowing. Differences are both semantical and syntactical. On the one hand, the CEP environment is characterized by an incoming and infinite stream of events. From there, a configurable, but constant sample (e.g., window) builds a sketch [START_REF] Garofalakis | Sketch-based geometric monitoring of distributed stream queries[END_REF] where aggregations are derived. On the other hand, SQL WF are computed over finite sets built from SQL relations. While the former windows are fixed and the data moves through, in the latter, the data is fixed and the window performs the movement. Moreover each approach considers distinct SQL keywords (e.g., OVER, RETAIN) and subsequent syntax.

Conclusion

WF with ranking analytical functions are required to have full disclosure of a given logical partition. Data partitioning is required to enable systems to scale, but harms data locality, which poses added difficulties when trying to parallelize these functions.

In this paper we motivate and validate how similarity between partitions can be used to promote efficient data forwarding among instances of a DQE. We introduced an algorithm to choose whether to batch or to hash forward rows between such instances by understanding how the similarity measure between distinct partitions of a DQE can be used towards the effectiveness of the WF environment.

The WF environment changes how analytical functions are computed, requiring specific implementation details for each functions. We therefore plan to leverage such parallelization opportunities to other analytical functions.

Fig. 1 :

 1 Fig. 1: Example of SQL query with WF.

Fig. 2 :

 2 Fig. 2: Stages of the Window operator: partitioning (1), ordering (2) and framing (3).

Fig. 3 :

 3 Fig. 3: WF query as: select analytical function() OVER (PARTITION BY A ORDER BY D) FROM table. (a) WF where the partition by clause generated 1 partitions. (b) Cumulative (sum) analytical function over WF in (a). (c) Ranking (rank) analytical function over WF in (a)

Fig. 4 :

 4 Fig. 4: Distributed Query Plan for Ranking WF. Round boxes represent individual stages of the WF environment. Arrows represent data flow in a process or over the network. PB and OB respectively represent Partition By and Order By attributes. H describes a statistical histogram. Numbers represent process execution order.shuffle mechanism[START_REF]Apache: The apache derby project[END_REF]. Afterwards, data is sorted according to the partitioning and ordering clauses (3), and results are submitted to the rank function[START_REF] Cao | Optimization of analytic window functions[END_REF]. At this stage, each computing node holds partial results from each logical partition. The results from each logical partition are then reunited in a single location (5) before being delivered to the client[START_REF] Coporation | Transact-SQL[END_REF]. Ranking aggregation algorithms are dependent on having full disclosure of the entire logical partitions. If the first shuffling stage (2) is not performed, the partial results in each partition will produce incorrect results. That is, if members of a logical partition are processed in the same node where they are stored (therefore in distinct DQE instances), the partial aggregation results produced will not be able to sort the entire logical partition. Thus, when the partial results of logical partitions are merged in the final result-set (6), they will need to be entirely recomputed. By considering the first shuffle stage (2), the results produced by stage (3) in each logical partition are globally correct since, independence from logical partitions ensures inter-partition parallelism, allowing computation to be distributed through several computing locations. The partition strategy considered depends on the mandatory argument of the partitioning clause.It is thus impractical to adjust the table splitting of the workload to a specific partitioning clause, since the ideal configuration may change with each query. Moreover, the environment allows the use of expressions as the arguments of the inner clauses, posing an extra hurdle to this abstraction.

Fig. 5 :

 5 Fig. 5: Shuffling instances partitioned by ol w. In WF context, they were partitioned by ol d and Ordered by ol num. The DQE instances will use the network to combine partitions during execution time. Instance w 1 will hold partitions ol d = 1, instance w 2 will hold partitions ol d = 2 and ol d = 3, respectively.

Fig. 6 :

 6 Fig. 6: Number of partitions per attribute (top) and the average number of elements per partition/attribute (bottom). The horizontal axis represents the attribute index. The vertical axis quantifies each measure in logarithmic scale. The attribute considered for placement key (PK) is shown in black and the candidates for WF Partition By key (PBK) are shown in dark gray.

Fig. 7 :

 7 Fig. 7: Similarity between attributes in two data nodes. Horizontal axis represents the attribute index. Vertical axis represents the Similarity measure in logarithmic scale.

Fig. 8 :

 8 Fig. 8: Bandwidth (outbound) registered during shuffling between instances.

Table 1 :

 1 Total Bandwidth (sent) and execution time registered for each configuration.

		2 nodes	4 nodes	8 nodes
	Baseline (MB)	1,132.45	4,172.59	7,237.56
	similarity (MB)	2,365.34	1,695.24	991.72
	Bandwidth Gain (x)	-0.48	2.46	7.30
			Shuffle time
	Baseline (sec)	149	172	170
	similarity (sec)	226	114	65
	Speed up (x)	-0.48	1.55	2.61

Typically, the use of this clause is restricted to numeric types.

Acknowledgments

The research leading to these results was part-funded by (1) the European Union's Horizon 2020 -The EU Framework Programme for Research and Innovation 2014-2020, under grant agreement No. 732051; (1) Project TEC4Growth -Pervasive Intelligence, Enhancers and Proofs of Concept with Industrial Impact/NORTE-01-0145-FEDER-000020 is financed by the North Portugal Regional Operational Programme (NORTE 2020), under the PORTUGAL 2020 Partnership Agreement, and through the European Regional Development Fund (ERDF) and by (1) the ERDF -European Regional Development Fund through the Operational Programme for Competitiveness and Internationalisation -COMPETE 2020 Programme within project POCI-01-0145-FEDER-006961, and by National Funds through the FCT -Fundac ¸ão para a Ciência e a Tecnologia (Portuguese Foundation for Science and Technology) as part of project UID/EEA/50014/2013.