Massively Scalable Data Warehouses with Performance Predictability

Abstract

Data Warehouses (DW) são ferramentas fundamentais no apoio ao processo de tomada de decisão, e que lidam com grandes volumes de dados cada vez maiores, que normalmente são armazenados usando o modelo em estrela (star schema). No entanto, o resultado das pesquisas e análises deve estar disponível em tempo útil. Contudo, como a complexidade das pesquisas que são submetidas é cada vez maior, com padrões de pesquisa imprevisíveis (ad-hoc), e devido ao aumento do número de pesquisas que são submetidas e executadas simultaneamente, provoca que o tempo de execução das pesquisas seja imprevisível. Mercados concorrenciais requerem que os resultados sejam disponibilizados em tempo útil para ajudar o processo de tomada de decisão. Isto não é apenas uma questão de obter resultados rápidos, mas de garantir que os resultados estarão disponíveis antes das decisões serem tomadas. Estratégias de pré-computação de pesquisas podem ajudar na obtenção de resultados mais rápidos, no entanto a sua utilização é limitada apenas a pesquisas com padrões conhecidos (planeados). Contudo, as consultas com padrões de pesquisa imprevisíveis (ad-hoc) são executadas sem quaisquer garantias de execução de tempo. São vários os fatores que influenciam a capacidade da DW fornecer resultados às pesquisas em tempo útil, tais como a complexidade da pesquisa (seletividade, número de tabelas que necessitam ser relacionadas, os algoritmos de junção e o tamanho das tabelas), a heterogeneidade e a capacidade da infraestrutura de processamento, incluindo a velocidade de leitura de disco, e à memória disponível para efetuar a junção das tabelas. O aumento do volume de dados e do número de pesquisas que estão a ser simultaneamente executadas também influenciam a capacidade do sistema em fornecer tempos de execução previsíveis. Apesar do tempo e esforço despendido para definir infraestruturas de processamento paralelo com capacidade para lidar com o aumento do volume de dados, e melhorar o tempo de execução das pesquisas, estas não permitem garantir a disponibilização atempada dos resultados, particularmente para as pesquisas ad-hoc. O tempo de execução de pesquisas com padrões conhecidos pode ser otimizado através de um conjunto de estratégias e mecanismos auxiliares, tais como a utilização de vistas materializadas e indexes. No entanto, para consultas ad-hoc, tais mecanismos não são uma solução. A imprevisibilidade do padrão de pesquisas origina tempos de execução imprevisíveis, que podem ser incompatíveis com os requisitos de negócio. Além disso, para muitos negócios, o crescente volume de dados condiciona ainda mais a capacidade da infraestrutura de processamento de fornecer resultados em tempo útil. Como consequência, os departamentos de TI estão constantemente atualizando a infraestrutura de processamento com a espectativa de que esta seja capaz de processar atempadamente as pesquisas, mas sem nenhuma garantia de que o consiga fazer. Não existe um método concreto que permita definir os requisitos mínimos de hardware que permita a execução atempada das pesquisas. Esta dissertação propõe uma arquitetura de Data Warehouse escalável com capacidade de lidar com grandes volumes de dados e de fornecer resultados em tempo útil, mesmo quando um grande número de pesquisas estão a ser simultaneamente executadas. A capacidade de fornecer resultados em tempo útil não é apenas uma questão de desempenho, mas uma questão de ser capaz de retornar atempadamente os resultados às pesquisas, quando esperado, de acordo com a natureza da análise e das decisões do negócio. O conceito de execução atempada (obtenção de resultados em tempo útil) é introduzido, e são propostos mecanismos que permitem fornecer garantias de execução atempada, sem no entanto descurar os requisitos de previsibilidade do tempo de execução das pesquisas e de latência mínima (frescura dos dados - freshness). A complexidade da execução de uma pesquisa é influenciada por diversos fatores, tais como a seletividade da pesquisa, o tamanho das tabelas, o número de junções e os algoritmos de junção. O volume de dados e memória disponível para junções, influenciam tanto a ordem de junção bem como o algoritmo de junção utilizado, resultando em custos de execução imprevisíveis. A necessidade de juntar as tabelas de dimensão com a tabela de factos advém do modelo em estrela (star-schema). O volume de dados é outro fator de imprevisibilidade, não sendo possível determinar com precisão o impacto do aumento do volume de dados no tempo de execução das pesquisas. Para lidar com estes fatores de imprevisibilidade relacionados com a junção de tabelas, propusemos o modelo de dados desnormalizado, chamado ONE. Neste modelo, os dados da tabela de factos, assim como os correspondentes dados das tabelas de dimensão, são fisicamente guardados numa única tabela desnormalizada, contendo todos os atributos das tabelas. O modelo de dados ONE requer mais espaço para guardar os dados, no entanto o modelo de processamento é mais simples e com tempos de execução previsíveis. Com o modelo de dados ONE, a tabela desnormalizada é particionada em fragmentos de dados mais pequenos e distribuídos pelos nós da infraestrutura para processamento paralelo, obtendo-se um aumento de desempenho. ONE possibilita uma escalabilidade quase ilimitada, uma vez que a totalidade dos dados (dos factos e das dimensões), e não apenas da tabela de factos, é linearmente dividida pelos nós da infraestrutura de processamento (com η nós homogéneos, cada nó conterá 1/η dos dados). Portanto, e uma vez que a adição de novos nós à infraestrutura de processamento não requer a replicação das dimensões, o modelo ONE oferece escalabilidade massiva de dados. Ao garantir uma distribuição linear de todos os dados, e não apenas os dados da tabela de fatos, o tempo de execução das pesquisas é melhorado proporcionalmente à redução do volume de dados em cada nó. Além disso, e porque os dados estão desnormalizados, o processamento das pesquisas é bastante simplificado e previsível, pois fica reduzido às operações de filtragem e de agregação dos dados. Como consequência, são reduzidos os requisitos da infraestrutura de processamento. Por norma, quando uma pesquisa é submetida não existe uma noção clara de quanto tempo irá demorar e se o resultado será obtido antes da tomada de decisão. Definimos o conceito de execução em tempo útil (right-time) como a capacidade de executar pesquisas de modo que os resultados estejam disponíveis antes da tomada de decisão (execução atempada), antes dum determinado objetivo temporal. O objetivo não é obter execuções mais rápidas, mas sim garantir que os resultados estarão disponíveis quando esperado. São propostos mecanismos que permitem fornecer previsibilidade de tempo de execução e garantias de execução atempada de pesquisas que tenham objetivos temporais. Como as pesquisas podem ter objetivos temporais diferentes do oferecido pela atual infraestrutura de processamento, propusemos um modelo de processamento chamado TEEPA (Timely Execution with Elastic Parallel Architecture), que toma em consideração os objetivos temporais das pesquisas para ajustar e rebalancear a infraestrutura de processamento de modo a que estes sejam garantidos. Quando a infraestrutura atual não consegue executar atempadamente as pesquisas, são adicionados mais nós de processamento e o volume de dados é redistribuído entre eles. Em cada nó, TEEPA monitora continuamente a execução da pesquisa, o volume de dados alocado, e a taxa de transferência IO, para determinar se as pesquisas podem ser atempadamente executadas. Como os nós de processamento podem ser heterogéneos, TEEPA toma em conta as suas capacidades de IO para determinar quantos nós são necessários e como deve ser efetuada a redistribuição dos dados. O volume de dados alocado em cada nó é ajustado em função do volume total (número total de registos), do tamanho do registo e da taxa de transferência de cada nó. Deste modo, a nós mais rápidos são atribuídos maiores volumes de dados. O processo de seleção e integração de novos nós de processamento e posterior rebalanceamento e reequilíbrio dos dados é executado até que os objetivos temporais sejam atingidos. Por outro lado, cada vez mais há a necessidade de analisar dados obtidos quase em tempo real, com mínima latência e frescura (freshness), o que requer que os dados sejam carregados mais frequentemente, à medida que são registados. Contudo, tipicamente as DW são refrescadas periodicamente com conjuntos de registos (batch), de modo a reduzir os custos de carregamento e os custos relacionados com o refrescamento de estruturas auxiliares, como índices e vistas materializadas. Sistemas de base de dados em memória minimizam estes custos, e possibilitam que os dados sejam carregados mais frequentemente. Contudo, a memória é finita e é insuficiente para conter a totalidades dos dados. De modo a oferecer latência mínima, definimos um modelo de processamento paralelo em que os dados são divididos em duas partes distintas: os dados antigos são guardados no modelo de dados ONE, ao qual chamámos Od, e os dados mais recentes são guardados em memória num modelo em estrela, designado de Os. Os dados podem ser carregados com maior frequência para Os, reduzindo assim a sua latência, e são aí mantidos enquanto existir memória disponível. Quando for necessário, por exemplo quando for necessário libertar memória para guardar novos dados, os dados mais antigos existentes em Os são movidos para Od. A utilização dum modelo hibrido, composto por Od e Os, permite que as DW existentes, que utilizam o modelo em estrela, possam ser migradas diretamente para este modelo com mínimo impacto ao nível dos processos de extração, transformação e carregamento dos dados (ETL). Na perspetiva do utilizador e das aplicações, este modelo hibrido oferece uma visão lógica dos dados num modelo em estrela, por forma a permitir uma fácil integração com aplicações e processos de carregamentos existentes, e a oferecer as vantagens do modelo em estrela, nomeadamente ao nível de usabilidade e facilidade de utilização. Uma camada de abstração gere a consistência de dados e processamento entre as duas componentes (Os e Od), incluindo a reescrita das pesquisas de modo a processar os dados que se encontram em cada uma das componentes. São também propostos mecanismos que oferecem garantias de execução atempada de pesquisas, mesmo quando um grande número de pesquisas está sendo processado simultaneamente. Infraestruturas paralelas podem minimizar esta questão, no entanto a sua escalabilidade é limitada pelo modelo de execução dos sistemas de bases de dados relacionais, onde cada pesquisa é processada individualmente e compete com as outras pelos recursos (IO, CPU, memória, …). É proposto um modelo de processamento de pesquisas, chamado SPIN, que analisa as pesquisas submetidas e, sempre que possível, efetua a partilha de dados e processamento entre elas, e assim consegue oferecer tempos de execução mais rápidos e previsíveis. SPIN utiliza o modelo de dados ONE, mas considera a tabela como sendo circular, isto é, uma tabela que é lida continuamente de uma forma circular. Enquanto existirem pesquisas a serem executadas, os dados são lidos sequencialmente e quando chega ao fim da tabela, recomeça a ler os dados desde o início da tabela. À medida que os dados são lidos, estes são colocados sequencialmente numa janela deslizante em memória (base pipeline), para serem partilhados pelas várias pesquisas. Cada pesquisa processa todos os registos da tabela, no entanto a leitura e o processamento não começa no registo número 1 da tabela, mas sim no primeiro registo da janela deslizante (início lógico). Os restantes registos são processados à medida que forem lidos e colocados na janela deslizante, até que o próximo registo a ser processado seja o do início lógico, isto é, após um ciclo completo. O custo da leitura dos dados é constante e partilhado por todas as pesquisas. Deste modo, a submissão de novas pesquisas não introduz custos adicionais ao nível da leitura de dados. O tempo de execução das pesquisas é influenciado apenas pela complexidade e número dos filtros (restrições) das pesquisas e pelo custo das agregações e ordenações dos dados. SPIN partilha dados e processamento entre pesquisas, combinando filtros e computações comuns a várias pesquisas num único fluxo (ramo) de processamento. Os vários ramos (branches) são sequencialmente conectados, formando uma estrutura em árvore que denominámos de WPtree (Workload Processing Tree), que tem como raiz o base pipeline. Quando uma pesquisa é submetida, se existir um ramo de processamento com predicados comuns aos da pesquisa, a pesquisa é encadeada como um novo ramo desse ramo comum, e são removidos os respetivos predicados da pesquisa. Se não existir um ramo com predicados comuns, a pesquisa é encadeada como um novo ramo do base pipeline. Deste modo, reduz-se o volume de dados que está em memória para processamento, bem como o custo de processamento dos predicados. A árvore de processamento é continuamente monitorizada, e quando necessário, um optimizador reorganiza dinamicamente o número e a ordem dos ramos. Sempre que possível, uma pesquisa é processada através da combinação dos resultados que estão a ser processados por outros ramos, e deste modo simplificando e reduzindo o volume de dados que a pesquisa tem que processar. Como os registos são lidos e processados pela mesma ordem, enquanto os dados não forem alterados, o resultado da avaliação dos predicados de cada registo é o igual ao da última vez que foi avaliado. De modo a evitar o custo da avaliação de registos anteriormente avaliados, e que não foram alterados, é proposta uma extensão ao modelo de processamento SPIN que utiliza uma abordagem de processamento baseada em bitsets (estruturas similares aos índices bitmaps). Um bitset é construído para cada ramo com o resultado da avaliação dos seus predicados, sendo o resultado de cada registo guardado na correspondente posição do bitset. Após o bitset estar completo, a posterior avaliação desses predicados pode ser substituído por uma simples verificação no bitset. Os bitsets têm um tamanho reduzido e são guardados em memória, de modo a evitar a introdução de custos adicionais ao nível de IO. Bitsets são particularmente relevantes para predicados complexos e com elevado custo de processamento, sendo criados e removidos dinamicamente de acordo com uma política de retenção, que toma em consideração vários aspetos, tais como a memória disponível, cardinalidade, e o custo da avaliação dos predicados. Através da análise do conjunto sequencial de ramos (path) de uma pesquisa, e dos custos de processamento de cada ramo, é possível estimar, com elevada precisão, o tempo de execução da pesquisa, mesmo quando existe um grande número de pesquisas a serem executadas simultaneamente. Para satisfazer pesquisas com objetivos temporais mais exigentes, é proposto um mecanismo de processamento, denominado CARROUSEL, que além de redistribuir e/ou replicar fragmentos dos dados pelos vários nós de processamento, redistribui também o processamento das pesquisas e dos ramos pelos nós. Tomando em consideração os bitsets existentes, é possível determinar quais os fragmentos de dados que cada pesquisa necessita processar e deste modo reduzir o custo de processamento através da ativação/desativação dinâmica dos ramos, consoante os fragmentos que estão nesse momento em memória. É possível terminar antecipadamente a execução de uma pesquisa, antes do término do ciclo. CARROUSEL é um processador flexível de fragmentos que utiliza um conjunto de nós inativos, ou nós que estão a executar pesquisas com objetivos temporais menos exigentes, para processar em paralelo alguns dos fragmentos de dados requeridos por pesquisas com objetivos temporais mais exigentes. Ao reduzir-se o volume de dados a ser processado por cada nó consegue-se tempos de execução mais rápidos. Alternativamente, alguns dos ramos de processamentos podem ser redistribuídos para outros nós com réplicas dos fragmentos. A execução da pesquisa termina quando todos os registos foram processados. No entanto, como os dados estão continuamente a ser lidos e à medida que são processados é recolhida informação relevante sobre os dados que existem em cada fragmento. Esta informação é relevante de modo a decidir como o balanceamento dos fragmentos e dos ramos a processar devem ser redistribuídos pelos nós por forma a reduzir custos de processamento e tempos de execução.

Data Warehouse (DW) systems are a fundamental tool for the decision-making process, have to deal with increasingly large data volumes, which is typically stored in as a star-schema model. The query workload is also more demanding, involving more complex, ad-hoc and unpredictable query patterns, with more simultaneous queries being submitted and executed concurrently. Modern competitive markets require decisions to be taken in a timely fashion. It is not just a matter of delivering fast analysis, but also of guaranteeing that they will be available before business-decisions are made. Moreover, the data volumes produced by data intensive industries are continuously increasing, stressing the processing infrastructure ability to provide such timely requirements even further. As a consequence, IT departments are continuously upgrading the processing infrastructure with the objective to hopefully the newer architecture will be able to deliver query results within the required time frame, but without any guarantees that it will be able to do so. There’s no concrete method to define the minimal hardware requirements to deliver timely query results. Several factors influence the ability of the DW infrastructure to provide timely results to queries, such as the query execution complexity (query selectivity, number of relations that have to be joined, the joins algorithms and the relations’ size), the heterogeneity and capabilities of the processing infrastructure, including IO throughput, and the memory available to process joins and the implementation of the join algorithms). Larger data volumes and concurrent query loads; concurrent queries that are executing simultaneously also influence the system ability to provide predictable execution times. In spite of all the time and effort to come up with a parallel infrastructure to handle such increase in data volume and to improve query execution time, it may be insufficient to provide timely execution queries, particularly for ad-hoc queries. The performance of well-known queries can be tuned through a set of auxiliary strategies and mechanisms, such as materialized views and index tuning. However, for ad-hoc queries, such mechanisms are not an alternative solution. The query patterns unpredictability result in unpredictable query execution times, which may be incompatible with business requirements. Data volumes produced by data intensive industries are continuously increasing, stressing the ability of the processing infrastructure to provide such timely answers even further. As a consequence, IT departments are continuously upgrading the processing infrastructure with the objective to deliver query results within the required time frame, but without any guarantees that it will be able to do so. There’s no concrete method to define the minimal hardware requirements to deliver timely query results. This dissertation proposes a data warehousing architecture that provides scalability and timely results for massive data volumes. The architecture is able to do this even in the presence of a large number of concurrent queries, and it is able to meet near real-time requirements. The ability to provide timely results is not just a performance issue (high throughput), but also a matter of returning query results when expected, according to the nature of the analysis and the business decisions. Query execution complexity is highly influenced by the number of relations that have to be joined together, the relations’ size and the query selection predicates (selectivity), influencing the data volume that has to be read from storage and joined. This data volume and the memory available for joins, influence both the join order and the used join algorithms. These unpredictable costs related to joining the fact table with dimensions relations arise from the star-schema model organization. The data volume is another factor of unpredictability, since there’s no simple and accurate method to determine the impact of larger data volumes in query execution time. To handle the unpredictability factors related to joining relations, we proposed the ONE data model, where the fact table and data from corresponding dimensions are physically stored into a single de-normalized relation, without primary and foreign keys, containing all the attributes from both fact and dimension tables. ONE trades-off storage space for a more simpler and predictable processing model. To provide horizontal scalability, we partitioned the de-normalized ONE relation into data fragments and distribute them among a set of processing nodes for parallel processing, yielding improved performance speedup. ONE delivers unlimited data scalability, since the whole data (fact and dimensions), and not just the fact table, is linearly partitioned among nodes (with η nodes, each will have 1/η of the ONE node). Therefore, since the addition of more nodes to the processing infrastructure does not require additional data replication of dimensions, ONE provides massive data scalability. By ensuring a linear distribution of the whole data, and not just the fact table, query execution time is improved proportionally to the data volume in each node. Moreover, since data in each node is already joined and thus query processing does not involve the execution of costly join algorithms, the speedup in each node is enhanced (almost) linearly as a function of the data volume that it has to process. By de-normalizing the data, we also decrease the nodes’ requirements, in what concerns physical memory (needed for processing joins), and query processing tasks, since the join processing tasks that were repeatedly (over and over) processed are removed. The remaining tasks, such as filtering and aggregations, have minimum memory and processing requirements. Only group by aggregations and sorting have memory requirements. The concept of timely results (right-time execution) is introduced, and we propose mechanisms to provide right-time guarantees while meeting runtime predictability and freshness requirements. The ability to provide right-time data analysis is gaining increasing importance, with more and more operational decisions being made using data analysis from the DW. The predictability of the query execution tasks is particularly relevant for providing right-time or real-time data analysis. We define right-time as the ability to deliver query results in a timely manner, before they are required. The aim is not to provide the fastest answers, but to guarantee that the answers will be available when expected and needed. We proposed a Timely Execution with Elastic Parallel Architecture (TEEPA) which takes into consideration the query time targets to adjust and rebalancing the processing infrastructure and thus providing right-time guarantees. When the current deployment is unable to deliver the time targets, it adds more processing nodes and redistributes the data volumes among them. TEEPA continuously monitors the local query execution, the IO throughput and the data volume allocated to each processing node, to determine if the system is able to satisfy the user specified time targets. TEEPA was designed to handle heterogeneous nodes and thus it takes into account their IO capabilities when performing the necessary data rebalancing tasks. The data volume allocated to each node is adjusted as a function of the whole data load (total number of tuples), the tuple size and the node’ sequential scan throughput, with larger data volumes allocated to faster processing nodes. The node allocation (selection and integration of newer nodes) and data rebalancing tasks are continuously executed until the time targets can be assured. There’s an increasing demand for data analyses over near real-time data, with low latency and minimum freshness, which requires data to be loaded more frequently or loaded in a row-by-row fashion. However, traditionally DWs are periodically refreshed in batches, to reduce IO loading costs and costs related to the refreshing indexes and pre-computed aggregation data structures. Main memory DBMS eliminate IO costs and thus can handle higher data loading frequencies. However, physical memory is limited in size and cannot typically hold the whole tables and structures. To provide freshness guarantees, the proposed architecture combines a parallel ONE deployment with an in-memory star-schema model holding recent data. The in-memory part (Os) maintains the recently loaded data, to allow the execution of real-time analyses. By using a star-schema model in Os, existing DW applications can be easily replaced and integrated with the architecture without the need to recreate the existing ETL tasks. Data is loaded into the in-memory Os and remains there for real-time processing while there’s memory available, so that the most recent data is held in the star-schema. When the physical memory is exhausted, the data in Os stored in the star-schema model is moved to Od in the ONE data model. From the user perspective and data presentation, the architecture offers a logical star-schema model view of the data, in order to provide easy integration with existing applications and because the model has advantages in what concerns users understanding and usability. A logical to physical layer manages data and processing consistency between models, including the necessary query rewriting for querying the data stored in each part, and merging of results. Finally we present the mechanisms of the architecture that allow it to still guarantee right-time execution in the presence of huge concurrent query loads. Modern DWs also suffers from workload scalability limitations, with more and more queries (in particular ad-hoc) being concurrently submitted. Larger parallel infrastructures can reduce this limitation, but its scalability is constrained by the query-at-time execution model of custom RDBMs, where each query is individually processed, competing for resources (IO, CPU, memory,…) and accessing the common base data, without data and processing sharing considerations. We propose SPIN, a data and processing sharing model that delivers predictable execution times for concurrent queries and overcomes the memory and scalability limitations of existing approaches. SPIN views the ONE relation in a node, as a logical circular relation, i.e. a relation that is constantly scanned in a circular fashion. When the end is reached, it continues scanning from the beginning, while there are queries running. Each query process all the required tuples of relation ONE, but the scanning and the query processing does not starts from the same first physical row. As the relation is read in a circular fashion, the first logical row is the one that already is cached in memory. The remaining tuples of the query are processed as they are being read from storage until the first logical row is reached. Data is read from storage and placed into an in–memory pipeline to be shared by all running concurrent queries. IO reading cost is constant and is shared between running queries. Therefore, the submission of additional queries does not incur in additional IO costs and joins operations. The execution times of concurrent queries are influenced by the number and complexity of the query constraints (filtering) and the cost of aggregations. To provide massive workload scalability it shares data and processing among queries, by combining the running queries in logical query branches for filtering clauses and by extensive reuse of common computations and aggregation operations. It analyses the query predicates, and if exists a logical branch in the current workload processing tree with common predicates it is registered in that logical branch, and the corresponding query predicates are removed. Otherwise, if do not exists a logical branch that meet the query predicates, it is registered as a new logical branch of the base data pipeline. This enhances processing sharing, and reduces the number of filtering conditions. The architecture has a branch optimizer that is continuously adjusting the number and order of the existing branches, and reorganizing them as required. Whenever possible, a query can merge and combine the results that are being processed by other branches, and thus simplifying and reducing the data volume that the query branch has to filter and to process. Since tuples flow using the same reading order, if data doesn’t change, the evaluation of the branch predicates against every tuple that flows along the branch will not change. The result of predicate evaluation will be the same as the last time it was evaluated. To avoid subsequent evaluation of unchanged data tuples, we extended the SPIN approach with a bitset processing approach. A branch bitset (bitmap) is built according to the branch’ predicates, where each bit represents the boolean result of the predicate evaluation (true/false) applied to a corresponding tuple index. Future evaluations of the tuple can take advantage of the existence of this bitset, since the selection operator that evaluates the predicate can be replaced by a fast lookup operator to the corresponding position in the bitset to gathers the result. Bitsets are small and reside in memory in order to avoid introducing overhead at IO level. This is particularly relevant for predicates with high evaluations costs. Through the analysis of the data path (branches) of queries, and the required computational costs of each branch, it is possible to determine high accurate estimations of query execution times. Therefore, predictable execution times can be given for massive workload scalability. Tighter right-time guarantees can be provided by extending the parallel infrastructure, and redistributing data among processing nodes, but also by redistributing queries, query processing and data branches between nodes holding replicated fragments. This is achieved by using two distinct approaches, a parallel fine-tuned fragment level processing, named CARROUSEL, and an early-end query processing mechanism. CARROUSEL is a flexible fragment processor that uses idle nodes, or nodes currently running less time-stricter queries, to process some of the fragments required by time-stricter queries, on behalf of the fragment node’s owner. By reducing the data volume to be processed by a node, it can provide faster execution times. Alternatively, it may distribute some logical data branches among nodes with replicated fragments, and thus reducing query processing. This is only possible with nodes with replicated data fragments. The execution of a query ends when all tuples of the data fragments are processed and the circular logical loop is completed. But as the system is continuously spinning, reading and processing over and over the same data, it collects insightful information regarding the data that is stored in each data fragment. For some logical data branches, this can be relevant to reduce memory and computational usage by using a postponed start (delaying the query execution until the first relevant fragment is loaded) and early-end approaches (detaching the query pipeline when all the relevant fragments for a query have been processed). This information is useful when the architecture needs to perform a data rebalancing process, with the rebalanced data being clustered according to logical branch predicates and stored as new data fragments.