A dataspace-based framework for OLAP analyses in a high-variety multistore

In a multistore, different data models may be used to represent and store data. This requires defining the basic database concepts in a way that abstracts from the single data models. Figure 4 provides a UML class diagram to describe the concepts that we use in this paper.

We use the term collection to refer to the container of data (i.e., what is known as table, column family, and collection in relational, wide-column, and document/key-value databases, respectively). Similarly, we introduce the term record to refer to the instances in a collection. Our notion of records perfectly corresponds to the tuples of a relational database, but the rows and documents of wide-column and document databases potentially correspond to multiple records. In fact, non-relational data models comply with the aggregate data modeling property, which enables the nesting of records within other records through the array data type. Thus, we do not consider documents and rows as a whole, but we separately model the records available at each nesting level.

From this point forward, we refer to primitive attributes or array attributes based on the respective type. Also, given an attribute a, we use name(a) and type(a) to, respectively, refer to its name and its type. If an attribute is nested within one or more array attributes, its name includes the dot-concatenation of the names of those array attributes. Finally, notice that: (i) arrays of primitive types are not considered for simplicity; (ii) attributes of object data type are not considered as they are simply containers of attributes for the same record (i.e., they entail the same expressiveness of primitive attributes); (iii) we exclude attributes of binary data types, whose values are uninterpretable without additional knowledge; (iv) to enable support to key-values stores (which support only two attributes, i.e., a string key and a binary value), we assume the value to contain interpretable strings; in particular, the name of the two attributes are inferred from the collection’s name (e.g., collection \(C_7\) in Fig. 3 has two attributes \(a_k\) and \(a_v\), where \(name(a_k)=\mathsf{invoiceId}\), \(name(a_v)=\mathsf{invoice}\), and \(type(a_k)=type(a_v)=\mathsf{string}\)).

For the sake of simplicity, given a record r, its schema (denoted with \(S_r\)) is the set of attributes directly available in r (i.e., without unnesting any array). If a record \(r'\) is nested within r, \(S_{r'}\) also includes \(key(S_{r})\); this is necessary to maintain the relationship between the schema of a nested record and the one of the parent record. Our schema definition provides a view of the records in first normal form, as it hides the denormalization due to the nesting of records and exposes the relationships between schemas at different nesting levels.

For the sake of simplicity, we assume all keys to be simple (i.e., not composite)². Also, it is reasonable to assume that all schemas (including those nested in arrays) have a key. We refer to \({\mathcal {S}}\) as the set of all schemas within the multistore.

Concerning the schemaless property of non-relational databases, we take into account every schema variation in a collection (i.e., if two records differ even for a single attribute, we model two separate schemas, each with its own set of attributes). Given our previous assumptions, collections in key-value stores are associated with a single schema (e.g., the schema of collection \(C_7\) is \(S = \{ \mathsf{invoideId}, \mathsf{invoice} \}\)).

Due to both schema variability and schema denormalization, attributes in different schemas may represent the same property. For example, in Fig. 5 different order line records use attributes with different names to indicate the quantity of product bought (i.e., quantity and qty, respectively). To resolve the different classes of heterogeneity and model the equivalence between different attributes of the dataspace we exploit mappings³.

For simplicity, we consider only simple mappings between two attributes. Since mappings are specified between attributes of different schemas, they reveal the relationship between such schemas. Consider two schemas \(S_i\) and \(S_j\).Mappings recognize that there is a semantic equivalence between two attributes in different schemas, thus we need to address all of them through a unique reference. This is the purpose of features.

Let attr(f) be the set of attributes represented by f (i.e., the representative attribute plus those derived from the mappings). Given a record r, the conflict-resolution function can be applied to \(r[a_i]\) and \(r[a_j]\) if \(\{ a_i, a_j \} \subseteq attr(f)\); we refer the reader to [16] for an indication about different methods to define conflict-resolution functions. Also, we remark that an attribute is always represented by one and only one feature; thus, for any two features \(f_i\) and \(f_j\), it is \(attr(f_i) \cap attr(f_j) =~\varnothing \). We use feat(a) to refer to the feature of an attribute a, name(f) to refer to the name of a feature, rep(f) to refer to the representative attribute of f, and rep(a) as short for rep(feat(a)).

Similarly to attributes, several schemas may be found to represent the same semantic concept (e.g., customers, orders). To hide such structural complexity, we introduce the concept of entities.

Ultimately, the dataspace is the data structure that puts together features and entities.

We say that \(E_i \xrightarrow {f} E_j\) if \(\exists ~ f \in E_i : \forall ~ a \in attr(f), a \in S_{E_i}\) it is \(S_{E_i} \xrightarrow {a} S_{E_j}, (S_{E_i}, S_{E_j}) \in ({\mathcal {S}}_{E_i},{\mathcal {S}}_{E_j})\). In other words, there is a many-to-one relationship from \(E_i\) to \(E_j\) on f if \(\forall ~S_{E_j} \in {\mathcal {S}}_{E_j}\) it is \(attr(f) \cap key(S_{E_j}) \ne \varnothing \) (i.e., the attributes of f are keys in the schemas of \(E_j\)) and \(\forall ~S_{E_i} \in {\mathcal {S}}_{E_i}\) it is \(attr(f) \cap key(S_{E_i}) = \varnothing \) (i.e., the attributes of f are not keys in the schemas of \(E_i\)). Similarly, we say that if \(\exists ~ f \in E_i : \forall ~ a \in attr(f), a \in S_{E_i}\) it is . Notice that we do not consider many-to-many relationships because schemas are inferred from physical implementations, where only many-to-one can be explicitly represented.

The iterative process to obtain, maintain, and use the dataspace is described in Fig. 8; the figure distinguishes the offline activities related to the management of the dataspace (in gray) from the online querying activity that relies on the dataspace (in white). Each step is described in the following.

Mapping definition. The goal of this step is to define mappings between the extracted schemas and attributes. This can be achieved in a semiautomatic manner, i.e., by combining the results of a schema matching algorithm [17] or tool (e.g., Coma 3.0 [18]) with knowledge manually provided by the user. In accordance with the pay-as-you-go philosophy, this methodology enables users to quickly reach the querying step; the mappings automatically defined by the algorithms are later refined by the user, as new insights are obtained through the querying of data.

Feature and entity recognition. This step is semiautomatic as well: based on Definitions 3.5 and 3.6 , both features and entities are automatically derivable from the mappings and the one-to-one relationships between schemas, respectively. Then, the user may refine the results by verifying whether or not the structural one-to-one relationships between schemas actually correspond to the same semantic concept. For instance, in our case study, Order schemas are in a one-to-one relationship with Invoice schemas, but they correspond to different semantic concepts. Similarly, given an entity E, \(\phi _E\) can be set manually or by running an automatic procedure that looks for matches between the key values across \({\mathcal {S}}_E\).

Querying. As soon as the dataspace is built, the user can exploit it to query the data; details on the querying step are given in Sect. 4. At any point in time, the user can go back to any of the previous steps to re-run some algorithm or to inject knowledge into the system.

The level of user intervention required in the semiautomatic activities of the offline phase is expected to decrease as users advance in the dataspace definition process: a core part of the dataspace will be stabilized after some iterations and it will be updated based on new user requirements (e.g., the exploration of attributes or schemas that had not been analyzed before), schema evolution, or the addition of new data sources. The cost of the update process remains constant over time, because (i) it only consists of updates at the metadata level, and (ii) it is safe to assume that existing constructs of the dataspace will not need to be redefined (if not to make corrections).

We rely on nested relational algebra (NRA) to define the execution plan of a query. Table 2 briefly explains each operator. With respect to traditional algebra, we introduce a new operator called merge ( ), i.e., an adaptation to our scenario of the full outerjoin-merge operator introduced in [19]. Its purpose is to replace the join operator (\(\bowtie \)) by addressing the extensional and intensional overlap between schemas. In particular, we consider the scenario in which records belonging to the same entity (e.g., Customer) can be partially overlapped, both in terms of instances (e.g., the same customer can be repeated across different schemas) and in terms of schemas (e.g., the name of the customer can be an attribute of two different schemas). We assume that the same is true when joining records from different entities (e.g., Customer and Order): the records can be partially overlapped (e.g., a customer may not have any orders, and an order may not be related to any customer), and so can be their schemas (e.g., the name of the customer can be used in the order’s schema as well).

We aim to keep as much information as possible when joining the records of two schemas, both from the extensional and the intensional points of view. The merge operator ( ) answers this need by (i) avoiding any loss of records, (ii) resolving mappings by providing output in terms of features instead of attributes, and (iii) resolving conflicts whenever necessary.

Building the execution plan of a query first requires identifying the entities that need to be accessed, which are not limited to those containing the features selected in the query. For instance, a query asking for the average price of the items ordered by a customer requires to access not only entities Customer and Orderline but also Order, even if no feature belonging to Order is mentioned in the query. Thus, we define the query graph as the subgraph of the entity graph that includes all and only the entities that need to be accessed to answer a certain query.

Condition (i) ensures that no unnecessary entity is accessed. Condition (ii) ensures that all attributes belonging to the features involved in the query are covered by the entities in \({\mathcal {E}}_q\). Condition (iii) entails the compliance of query q with the GPSJ semantics, that is, there exists an entity representing the events at the finest level of granularity (i.e., \(E^* \Rightarrow E^{'}\) indicates that a directed path exists from \(E^*\) to every other entity \(E^{'} \in {\mathcal {E}}_q\)). Many subgraphs could exist for a given query since many--to-one paths could exist, each associated with different semantics (e.g., an entity of sales could be associated with an entity of dates through the mappings on both date of sale and date of shipping). In this case, we rely on a user interaction to identify the adequate subgraph.

The query graph \(G^{{\mathcal {E}}}_q\) is the starting point to define the execution plan in NRA for query q, i.e., the query plan \(P_q\) (an example is shown in Fig. 10).

As shown in Fig. 10, the leaves of the query plan can be organized into entity plans, and the leaves of each entity plan can be organized into collection plans. In the following paragraphs, we describe the top-down decomposition of query, entity, and collection plans, and the procedure to obtain them from the query graph. Such a procedure embeds a series of optimization techniques, which we highlight in Sect. 4.3.

The distributed and multi-model nature of the multistore environment, coupled with the high-variety scenario covered in this paper, offers several opportunities for the optimization of query plans. The approach described so far already adopts a set of optimization techniques to produce a refined execution plan.Although some of the mentioned optimizations are not new to DBMSs and execution engines, their application in a complex multistore environment is not straightforward. Apache Spark (i.e., the one we use in our prototype) uses Catalyst to provide optimization techniques in query executions. However, Catalyst is not aware of the constraints that guarantee the correctness of the query plan and, ultimately, of the query result. As explained above, our heuristics preserves the inner structure of entity plans when reordering them within a query plan (and similarly for collection plans reordering within an entity plan). Since Catalyst has no notion of the internal organization of the query plan, its reordering strategy may swap operations that break the boundaries of collection or entity plans (e.g., a collection plan may be moved to a different entity plan), thus compromising the correctness of the result. For this reason, these optimization routines are directly defined within our approach.

Our reference architecture is a two-rack Big Data cluster of 18 Ubuntu machines with a minimum configuration of i7 8-core CPU @3.2GHz, 32GB RAM, and 6TB hard disk drives. Each machine runs the Cloudera Distribution for Apache Hadoop (CDH) 6.2.0. The multistore implementation relies on PostgreSQL, MongoDB, Cassandra, and Redis as relational, document-based, wide-column, and key-value DBMSs, respectively. PostgreSQL is installed on a single machine, while NoSQL stores are distributed across 15 machines. The algorithmic implementation of the approach is based on Apache Spark, i.e., one of the most used open-source execution frameworks for Apache Hadoop clusters; it provides connectors to most DBMSs, including those in our multistore.

Figure 11 provides an overview of our prototypical implementation from a functional and technological perspective. The main application modules (i.e., the query planner and the dataspace manager) are written in Scala, and they are coupled with an HTTP server that enables user interactions through REST APIs. The dataspace manager includes functionalities to build, update, and visualize the content of the dataspace (whose metadata are stored in the same PostgreSQL instance used for the data), while the query planner implements the algorithms (i.e., Algorithms 1 to 3) and the optimization techniques (see Sect. 4.3). Queries are formulated by relying on the SQL APIs exposed by Spark’s DataFrame abstraction; the new merge operator fits this abstraction: it is implemented as a full-outerjoin between two DataFrames, on top of which are applied custom User Defined Functions (UDFs) representing the conflict-resolution functions.

The query execution framework consists of 8 executors, each with 6 CPU cores and 8GB RAM⁵. The data is collected from the underlying DBMSs by pushing to the latter as much computation as possible; then, the Query execution framework runs in-memory computation to complete the execution of the query and obtain the final results, that are finally returned to the user.

To evaluate the approach in terms of scalability, we have implemented the multistore in four scale factors, i.e., 1, 10, 100, and 1000. The size of each collection in the different scale factors is reported in Table 3. We recall from Sect. 2 that there is a \(20\%\) overlap of customers between \(C_1\) and \(C_5\), and a \(60\%\) overlap of products between \(C_4\) and \(C_6\). The number of products is fixed in each implementation, together with the ratio of orders per customer (i.e., 15 on average) and the ratio of order lines per order (i.e., 5 on average); what scales is the number of customers and, consequently, the overall number of orders and order lines. Table 3 also reports the partitioning key of each collection; whereas partitioning (i.e., sharding) records is a necessity in distributed DBMSs, it also helps increasing query efficiency in presence of certain selection predicates in every DBMS. Notice that \(C_4\) in partitioned only on a customer attribute because orders and order lines are nested within customers.

As we recall from Sect. 3.3, the preparation of the dataspace is done in three steps.Ultimately, the dataspace’s metadata occupy less than 100 kB in every scale factor.

The first experiments are aimed at assessing the scalability of the system under different levels of variety in the data. In particular, we measure how the query planner and the merge operator perform by varying the number of schemas and the amount of overlapping records, respectively.

To evaluate the query planner, we build two synthetic collections of customer records, each with 50000 records; one is stored on the relational database with a single schema, the other on the document-based database with a varying number of schemas, from 1 to 10000. The latter is a borderline scenario, as (from our experience) collections with high variety rarely exceed the hundreds of schemas. Figure 12 shows execution times (averaged from 5 executions) of a query that merges and aggregates the data from both collections; we consider a single-core Spark instance of the middleware, so as to exclude variations due to parallelization. The results show that both query planning and execution are not affected by the number schemas, as minimum oscillations are observed. This is expected for the query execution, since resolving schema heterogeneity consists of low-impact operations such as renaming attributes’ names. As to query planning, even though the complexity of the procedure is linear with the number of schemas, the cardinality of the latter is not sufficient to impact the overall planning time. Ultimately, this proves a good efficiency of the query planner in handling high levels of schema heterogeneity.

As to the merge operator, we measure its performance under varying levels of record overlapping. Starting from the two previous collections of customer records, we remove schema heterogeneity and progressively increase the level of overlap between the records from 0% to 100%. The results are shown in Fig. 13; the execution times (averaged from 5 executions) correspond to the single merge operation (i.e., the two read operations are not considered). By increasing the level of overlap, the merge operation naturally returns a progressively lower amount of records; nonetheless, the performance of the merge operator is not influenced by this factor (the observed variations are minimal). This behavior is expected, as the complexity of the merge operation is the same as a full-outerjoin operation and the conflict-resolution functions are not computationally expensive.

The workload we devise consists of 48 GPSJ queries that vary in terms of group-by set strength, selection predicate selectivity, and the number of entities involved (i.e., the size of the query graph).This determines a total of 54 combinations; however, queries with no group-by set and no/weak selection predicates (i.e., non-analytical queries) are hardly applicable in large query graphs, where the cardinality of the result would be close to the size of the entire database. Thus, we exclude these two kinds of queries on the three largest query graphs, obtaining a total of 48 queries. The detailed list of queries is provided as Supplementary Information with the paper.

Execution times and scalability. The workload queries have been executed on the multistore against every scale factor. The execution time (always obtained as the average of 5 executions) mainly depends on the complexity of both the query and the dataset, but it is also affected by the way the computing resources have been allocated on the cluster. Big Data frameworks like Spark try to honor the locality principle, but they do not guarantee that the computation always happens on the same nodes; thus, execution times of the same computation may vary depending on the amount of data shuffling required when the locality principle is not met. Conversely, the time taken to build the execution plan (i.e., by running Algorithms 1–3) is affected by neither the query and dataset complexity (as shown in Sect. 5.2) nor the Big Data framework (as the implementation is centralized), and it always performs in sub-second times.

The query execution times (in seconds, together with the relative standard deviation (RSD)) are shown in Tables 4, 5, and 6; each table shows average times by, respectively, grouping the workload queries by group-set strength, selection predicate strength, and the number of entities in the query. Times increase as expected with the scale factor (especially evident when moving from SF 10 to 100), while selection predicates appear to have little effect. Indeed, the system can exploit local indexing and/or partitioning only on the collections on which the selection predicates are applied. The behavior under different group-by conditions is also different: execution times are faster in absence of group-by set because no data shuffling is required to carry out an aggregation; when the aggregation is necessary, the system performs slightly better if the group-by set is stronger, where fewer records are generated and shuffled. This is due to the usage of combining strategies that carry out map-side aggregation, thus shuffling less records for the reduce-side aggregation.

Local vs middleware computation. A second evaluation is made to compare the amount of computation assigned to the source against the one assigned to the middleware. For each query execution, we consider the local computation as the sum of the execution times of Spark’s tasks in charge of reading from the DBMSs, and middleware computation as the sum of the execution times of Spark’s remaining tasks⁶. The results are shown in Fig. 14. Interestingly, the percentage of computation demanded from the middleware decreases with the increase in the scale factor. In absolute terms, the local computation demanded from the sources scales linearly with the scale factor, while the middleware computation initially scales sublinearly (about 2x from SF 1 to SF 10, about 5x from SF 10 to SF 100). This is due to the middleware suffering the distributed framework’s overhead in handling low amounts of data in the smaller scale factors. Ultimately, we infer that relying on middleware for joining and merging records (which involves shuffling data on the network between different software tools) does not have a major impact, especially when the amount of data to be considered becomes larger.

Optimization impact. Finally, we measure the impact of our optimization techniques by selectively switching them off and verifying the execution times. We specifically focus on the schema plan grouping (SPG), merge sequence reordering (MSR), and column pruning (CP) optimizations. In this case, we obtain the measurements for each query and evaluate, on each scale factor, the average loss in percentage with respect to the execution with every optimization enabled. The results are shown in Table 7. While the contribution of CP is limited and erratic, SPG emerges as the optimization producing the most significant advantage. This is expected, as turning it off means issuing several queries on the same collections, which clearly has a major impact—especially with increasing scale factors, where the weight of the local computation is higher (as seen in the previous evaluation). MSR is also quite relevant; unlike SPG, MSR’s contribution is decreasing with the scale factor, since the weight of the middleware computation decreases as well; the only exception is in SF1, where the benefit of MSR in queries with low execution times is mitigated by the distributed framework’s overhead.

Adopting a pay-as-you-go approach entails that query answer quality depends on the number of defined mappings. In this subsection, we analyze how the results vary by selectively removing some of the mappings. Our goal is to quantify the impact of mappings in producing a correct result and to demonstrate the issues that would arise by adopting a system that does not entail a mechanism to solve schema heterogeneity and record overlapping.

Let \({\mathcal {D}}^*\) be the ground-truth dataspace (i.e., the one with all mappings identified); we consider three different scenarios, each represented by a different dataspace (i.e., \({\mathcal {D}}_1\), \({\mathcal {D}}_2\), \({\mathcal {D}}_3\)), where different types of mappings have been selectively removed with respect to \({\mathcal {D}}^*\) (we refer the reader to Table 1 for attributes’ and features’ definitions in our case study). Table 8 summarizes the characteristics of each scenario and measures the average quality degradation of those workload queries that are affected by the removal of the mappings. Inspired by [19], Table 9 evaluates the following.

Simple attributes. In \({\mathcal {D}}_1\) we consider a lack of mapping between attributes within the same entity; for instance, the attributes representing the OrderDate of the Order are not reconciled by a mapping, thus \(a_{29} \not \equiv a_{30}\). Failing to recognize this kind of mapping means that each attribute gets represented by a distinct feature (e.g., \(f^{'}_{9}\) in \(C_2\) and \(f^{''}_{9}\) in \(C_5\)) and must be queried separately. This scenario has an impact on the query results in terms of density (i.e., the percentage of non-null values), meaning that:

Simple attributes with record overlapping. In \({\mathcal {D}}_2\) we suppose that the same scenario in \({\mathcal {D}}_1\) applies to attributes of collections with overlapping records; for instance, the attributes representing the LastName of Customers are not reconciled by a mapping, thus \(a_{36} \not \equiv a_{37}\). Failing to recognize this kind of mapping means not only that (i) as in \({\mathcal {D}}_1\), each attribute gets represented by a distinct feature (e.g., \(f^{'}_{12}\) in \(C_1\) and \(f^{''}_{12}\) in \(C_5\)), but (ii) it also introduces a problem in terms of veracity of the results. When records are overlapping, any potential conflict (e.g., different last names found in different records of the same customer) are solved by the merge functions defined in the features in \({\mathcal {D}}^*\). In \({\mathcal {D}}_2\), having distinct features means that the respective attributes can be queried separately, but the obtained results cannot be easily merged. For instance, consider two queries that sum the TotalPrice by LastName, i.e., \(q' = (\{ f^{'}_{12} \}, \{ f_8, sum() \})\), and \(q'' = (\{ f^{''}_{12} \}, \{ f_8, sum() \})\); an excerpt of the queries’ results is shown in Table 9, together with the actual results from \({\mathcal {D}}^*\). Without record overlapping, the results from \(q'\) could have been summed to those from \(q''\) to obtain the ground truth values. This is not necessarily true in presence of record overlapping, because the function in \({\mathcal {D}}^*\) resolves conflicts in the last names before the aggregation and produces different results. In particular, we measured a \(\pm 111\%\) difference between the sums of total prices obtained in \({\mathcal {D}}^*\) and those obtained in \({\mathcal {D}}_2\) by summing the results of \(q'\) and \(q''\).

Key attributes. In \({\mathcal {D}}_3\) we consider a lack of mapping involving key attributes; for instance, \(a_{32}\) (i.e., the key of Customer in \(C_1\)) is not mapped to either \(a_{34}\) (i.e., the key of Customer in \(C_5\)), nor to \(a_{33}\) and \(a_{35}\) (i.e., the attributes in the Order referencing the key of the Customer). Failing to recognize this kind of mapping means that (i) as in \({\mathcal {D}}_1\) and \({\mathcal {D}}_2\), two features are created in \({\mathcal {D}}_3\) to represent the TaxId (e.g., \(f^{'}_{11}\) and \(f^{''}_{11}\)), but also that (ii) two separate entities are defined to represent customers (e.g., \(\mathsf{Customer}^{'}\) and \( \mathsf{Customer}^{''}\)), where only one of the two entities is actually linked to the Order. The main impact of this scenario on the query results is in terms of coverage [19] (i.e., the number of returned records), meaning that the records of the customer cannot be queried altogether. In particular:

The novel scenario considered in this paper is the one where schema heterogeneity and record overlapping prevent users from directly issuing analytical queries over a multistore. In this section, we compare with alternative approaches by analyzing how the latter would tackle the same problem.

Reconciled level materialization. This is the classic Data Warehouse approach: a fully reconciled schema is created and loaded in batch mode via an ETL procedure [25, 26]. Alternatively, a trigger-based approach can be adopted to feed the materialized view [27]. This solution favors the optimization of query time at the expense of making the system very rigid: (a) maintainability is affected, since every schema change entails an update of the ETL procedure; (b) the pay-as-you-go principle is compromised and a strong initial modeling effort is required; (c) materialized views and data sources are no more synchronized and the misalignment depends on how often the ETL procedure is executed. In our case study, the time to run a full materialization scales linearly with the scale factor and reaches up to 6 hours with SF 1000.

Data model transformation. This class of work suggests performing data model transformation to facilitate the access to data having heterogeneous structures. The common strategy consists in changing the underlying data model, usually from a non-relational to a relational data model. This kind of solution leads to the loss of the schemaless flexibility guaranteed in most NoSQL stores in favor of the use of conventional relational querying and storing techniques. In particular, custom transformations and mappings are typically defined to move data from one data model to the other [25, 26]. A mainstream approach widely used while dealing with heterogeneous XML databases is to transform documents into relation data model [30‐32]. Other alternatives suggest storing documents on the wide-column data model. For instance, MonetDB [33] uses specialized data encoding, join methods, and storage for managing documents encoded in XML on the wide-column data model. In [30], the authors use the document type definition, i.e., DTD, to flatten documents and map documents into relational tables. However, despite the advantages of using relational schema and the expressiveness power of relational operators, partitioning data into tables by attributes [32] affects the performance of the relational system. This is due to the need of performing multiple joins to reconstruct the initial data. Furthermore, users of these systems have to learn new schemas every time new data are inserted (or updated) because it is necessary to re-generate the relational views. Another line of work introduces data model transformation between NoSQL stores. In [34], the authors introduce a tool-based advisor with a cost model dedicated to data migration scenarios. However, this approach mainly focuses on optimizing the costs related to migrating data from one data model to another, and it does not consider cross-data model querying nor resolving the problem of schema heterogeneity within the same data model.

Schema Versioning This class of work identifies the different co-existing versions within one database and adds a transparent layer to query tables having different representations for their schema regardless of the version used to formulate queries. This line of work mainly targets relational databases and suggests changing the physical storage when a new version of the schema needs to be materialized. In [35] the authors introduced the Bi-directional Database Evolution Language BiDEL as a solution to automatically generate queries that match the different structures within a relational database. Therefore, the users formulate their queries regardless of the schema version. This solution does not address record overlapping and is designed to support schema versioning in relational databases (where the schema should be defined before loading the data), whereas NoSQL databases store the data without any prior data validation or structure verification. Recent work considers schema versioning in the context of NoSQL stores. In [36], the authors introduce forward and backward query rewriting for querying data with different versions. Furthermore, it is possible to have heterogeneity within the same database, e.g., different cardinally, and different structures. However, the approach is limited to solving heterogeneity within a single collection, it requires a history graph of schema evolutions to enable query rewriting (which is not necessarily available), and it does not address record overlapping.

Schema-independent querying. This class of work proposes solutions to overcome schema heterogeneity by enabling schema-independent querying; in particular, the common strategy is to rely on query rewriting techniques [37] to reformulate an input query into several derivations, thus overcoming schema heterogeneity. Most research work is designed in the context of relational databases, where heterogeneity is usually restricted to the lexical level. When it comes to the hierarchical nature of semi-structured data (XML, JSON documents), the problem of identifying similar nodes is insufficient to resolve the problem of querying documents with structural heterogeneity for instance. To this end, keyword querying has been adopted in the context of XML [38]. The process of answering a keyword query on XML data starts with the identification of the existence of the keywords within the documents without the need to know the underlying schemas. The problem is that the results do not consider heterogeneity in terms of nodes, but assume that if the keyword is found, no matter what its containing node is, the document is adequate and must be returned to the user. Recent research work introduced a transparent querying mechanism to enable querying for heterogeneous documents. In [39], the authors introduced novel querying mechanisms based on query rewriting techniques [36] where they overcome the problem of structural heterogeneity in document stores. Their contribution consists of generating a dictionary with different attributes and their corresponding paths. Then, a query reformulation engine enriches the initial user queries with all possible paths extracted from the dictionary. In the same direction, another research work [40] resolves the problem of querying heterogeneous documents by covering a broader class of heterogeneity. Thus, the authors resolve the problem of having heterogeneous attributes that are semantically equivalent but with a different naming convention, as highlighted in [41], using a set of schema mappings. However, the queries must be combined to retrieve data from different structures. Overall, we notice that most of the schema-independent querying approaches consider the heterogeneity problem inside one collection at a time for a particular data model only. Furthermore, the resolution of schema heterogeneity is usually limited to a given type of heterogeneity. For instance, structural, or semantic whereas more classes of heterogeneity could be identified in NoSQL stores. For instance, the same information could be represented using different data types, and transcoding functions are required to resolve this heterogeneity [42].

Schema inference. To assist the users while formulating their queries, several research efforts have been directed toward schema inference techniques. The idea is to provide users with an overview of the different elements present in the heterogeneous data, e.g., document stores, [43, 44]. This family of work was first introduced for inferring structures from semi-structured documents encoded in XML format. These papers aim to infer structures using regular expression rules from the different strings representing elements from XML documents to propose a generalized structure [45]. Both, JSON and XML are commonly used to encode nested data as documents. However, most of the solutions introduced to infer structures from documents encoded in XML could not be applied to documents encoded in JSON. Furthermore, other efforts were conducted to infer RDF data [46]. The problem with this class of work is none of these approaches is designed to deal with massive datasets whereas current applications are data intensive, and they are using JSON encoding.

In [41], the authors propose a framework to efficiently discover the existence of fields or sub-schemas inside the collection. To this end, the framework is built for managing a schema repository for JSON document stores. The proposed approach relies on a notion of JSON schema called skeleton, i.e., a tree representation describing the structures that frequently appear in a collection of heterogeneous documents. Thus, the skeleton may lack some paths that do exist in some of the documents because they do not appear often, and the generation of the skeleton will exclude them. Similarly, in [47] is proposed a schema profiling approach for document stores, where the goal is to expose the rules that drive the usage of different schemas to represent the same data. However, this approach is focused on providing insights into the users, but it does not provide any querying mechanism. In [48], a novel technique is defined to explain the schema variants within a collection in document stores. Therefore, the heterogeneity problem in this research work is detected when the same attribute is represented differently, e.g., different types, different locations inside documents. Therefore, the authors suggest using mappings to find out the different variations for a given attribute and ultimately build a multidimensional integrated view of the data to support OLAP queries. The main limitations of this approach are that it focuses on one collection at a time and that the query rewriting mechanism creates one query for every schema variation detected in such collection.

Overall, these works infer the implicit structures from heterogeneous data and provide the user with a high-level illustration regarding all or a subset of structures present inside the heterogeneous data. Schema inference techniques could help users to better understand the different underlying structures and to take the necessary measures and decisions during the application design phase. The limitation with such a logical view is that it requires a manual process to build the desired queries by including the desired attributes and all their possible navigational paths. In such approaches, the user is aware of data structures but is required to manage the heterogeneity. Furthermore, some proposals do not consider all structures and build an inferred schema on top of most used attributes, for instance, using some probability measures. Thus, queries could result in misleading results. Also, most of the proposals do not offer automatic support for structural evaluations and it is mandatory to regenerate the inference process which could affect the associated workloads and applications.

Multistore and polystores. In this part, we consider multistore and polystore systems providing integrated access and querying to several heterogeneous stores through a mediator layer. Systems like Teradata [49] or HadoopDB [50] propose to partition data between stores. Furthermore, they allow queries to access data shredded in the different stores and to move processing and/or data between stores. Such solutions require to co-locate stores within the same physical nodes to reduce the traffic overheads between nodes since data has to be moved to execute the queries, and different systems have to share each others’ partitioning strategies. More recent proposals ensure access to the data using either a novel unified querying language (e.g., SQL++ [51]) or by supporting both the query languages of the underlying stores and a unified querying language (e.g., Spark SQL [52]). In [53] authors leverage several databases and processing platforms, and they define a unified declarative processing interface to access and query heterogeneous data. Another alternative to accessing the data is to formulate several queries using the different underlying stores querying languages and to employ a middle-ware layer to merge and return the final results [54]. More recent proposals consider wider support of integrated systems; for instance, ESTOCADA [55] supports key-value, document, relational, and nested relational data stores. Overall, we notice that despite the efficient support of different data models, the proposed multistore and polystore systems do not support schema heterogeneity.

Multi-model systems. In contrast to multistore systems (where data is stored in different stores), multi-model systems offer a single database to store and manage different data models by offering an integrated system to guarantee large scale databases requirements in terms of storage, availability, and fault tolerance (e.g., OrientDB, http://​orientdb.​com/​orientdb/​). The concept of multi-model systems was earlier introduced in the literature with the ORDBMS systems (object-relational database management systems) offering support to object-oriented programming with relational databases [56]. Recent multi-model systems advocate the idea of reducing the task of combining partial results from different stores and thus suggest having an integrated database, which hides the heterogeneity in terms of data models by providing a declarative approach of querying multi-model data. Therefore, data model transformation can be carried out only when it is required. In [57], this philosophy is embraced to propose a multi-model approach to data warehousing.

Ultimately, multi-model systems excel in terms of data governance, management, and access. It is only required to maintain one system while taking advantage of several data models. However, existing systems are limited to a pre-defined set of data models, extending support to new data models is challenging, and (most importantly) they do not provide any mechanism to handle schema heterogeneity (e.g., reconciling the usage of different naming conventions for the same attribute) nor record overlapping.

Effectively supporting querying on a heterogeneous system with overlapping records requires the adoption of data fusion techniques [58]. The literature on this subject is very wide, thus we refer the reader to a recent survey [59]. Among the most important ones, we outline [19], where the authors propose a relational algebra operator (called full-outerjoin merge) to carry out data fusion while joining two tables—which is also the inspiration for the definition of the merge operator introduced in Sect. 4.1. Remarkably, related works in this area do not apply directly to a polyglot system; their scope is focused on the recognition and resolution of conflicts between records representing the same entity, but their application is mostly independent of the contextual storage and querying system. The related literature dealing with this problem is a building block of our approach, but it is not sufficient to address our complex multistore scenario on its own.

To the best of our knowledge, the only proposal that considers a scenario requiring data fusion in a polyglot system is QUEPA [60], where the authors present a polystore-based approach to support query augmentation. The idea is to let the user issue a query onto a single DBMS (using its native query language) and to augment query results with related information taken from the other DBMSs. The approach is meant to complement the other polystore systems that actually support cross-DBMS querying, and record linkage techniques are only used to find related instances in different DBMSs, but not to solve conflicts. Unlike [60], we (i) offer an integrated dataspace view over the whole multistore, (ii) enable cross-DBMS querying, and (iii) apply data fusion techniques at query time to solve conflicts in the data and return a polished result.