Clustering-based fragmentation and data replication for flexible query answering in distributed databases

As the basic data model, we consider the case of data stored in relational tables. The relational data model is still widely applied today although alternatives exist (like tree- or graph-structured data or data stored in a simple key-value format).

In this paper we focus on flexible query answering for conjunctive queries expressed as logical formulas. That is, we assume a logical language L consisting of a finite set of predicate symbols (denoting the table names; for example, Ill, Treat or P), a possibly infinite set dom of constant symbols (denoting the values in table cells; for example, Mary or a), and an infinite set of variables (x or y). A term is either a constant or a variable. The capital letter X denotes a vector of variables; if the order of variables in X does not matter, we identify X with the set of its variables and apply set operators - for example we write y ϵ X. We use the standard logical connectors conjunction ∧, disjunction ∨, negation - and material implication → and universal ∀ as well as existential ∃ quantifiers. An atom is a formula consisting of a single predicate symbol only; a literal is an atom (a “positive literal”) or a negation of an atom (a “negative literal”); a clause is a disjunction of atoms; a ground formula is one that contains no variables; the existential (universal) closure of a formula ϕ is written as ∃ϕ (∀ϕ) and denotes the closed formula obtained by binding all free variables of ϕ with the respective quantifier.

A query formula Q is a conjunction of literals with some variables X occurring freely (that is, not bound by variables); that is, Q ( X ) = L i 1 ∧ … ∧ L i n. By abuse of notation, we will also write L _ij ϵ Q when L _ij is a conjunct in formula Q. A query Q(X) is sent to a knowledge base Σ (a set of logical formulas) and then evaluated in Σ by a function a n s that returns a set of answers containing instantiations of the free variables (in other words, a set of formulas that are logically implied by Σ); as we focus on the generalization of queries, we assume the a n s function and an appropriate notion of logical truth given. A special case of a knowledge base can be a relational database with database tables as in Example 1.

To achieve fault tolerance, reliability and high availability, data in a distributed database system should be copied (that is, replicated) to different servers. Whenever one of the database servers fails, if it is too overloaded or geographically too far away from the requesting user, a data copy (that is, a replica) can be retrieved from one of the other servers.

The data replication problem (DRP; see [6]) is a formal description of the task of distributing copies of data records (that is, database fragments) among a set of servers in a distributed database system. The data replication problem is basically a Bin Packing Problem (BPP) in the following sense:

This BPP can be written as an integer linear program (ILP) as follows - where x _ik is a binary variable that denotes whether fragment/object i is placed in server/bin k; and y _k denotes that server/bin k is used (that is, is non-empty):

To explain, Equation 1 means that we want to minimize the number of servers/bins used; Equation 2 means that each object is assigned to exactly one bin; Equation 3 means that the capacity of each server is not exceeded; and the last two equations denote that the variables are binary - that is, the ILP is a so-called 0-1 linear program.

An extension of the basic BPP will be used to ensure that replicas will be placed on distinct servers: the Bin Packing with Conflicts (BPPC; [7]-[9]) problem allows constraints to be expressed on pairs of objects that should not be placed in the same bin. That is, one adds a conflict graph G=(V,E) where the node set V={1,…,n} corresponds to the set of objects. A binary edge e=(i,j) exists whenever the two incident nodes i and j must not be placed in the same bin; note that (i,j) is meant to be undirected and hence identical to (j,i). In the ILP representation, a further constraint is added to avoid conflicts in the placements.

Equation 9 ensures that no conflicting objects i and j are placed in the same bin k because otherwise the sum of the two x-variables x _ik and x _jk would be 2 and hence exceed y _k which is 1.

In this paper, we will extend the BPPC to ensure that a certain replication factor m for each fragment of the relational table is obeyed; that is, for each fragment stored at one server there are at least m - 1 other servers storing a copy of this fragment, too.

Let A be a relaxation attribute; let F be a table instance (a set of tuples); let C={c₁,…c _n } be a complete clustering of the active domain π _A (F) of A in F; let h e a d _i ϵc _i ; then, a set of fragments {F₁,…,F _n } (defined over the same attributes as F) is a clustering-based fragmentation if

We use and adapt an established approximation algorithm for clustering originally presented by Gonzalez [10]. Its original presentation relies on a notion of distance between any two values. It has a running time of O(k f) for clustering a set of k objects into f clusters. Each cluster is represented by one or more so-called head values; and each value is assigned to the cluster represented by a head with minimal distance to the value. In case the distance measure is metric (in particular, satisfies the triangular inequation), Gonzalez showed that the number of heads obtained by his algorithm is at most twice as much as the optimal number of heads (in other words, it is a 2-approximation of the optimal solution).

Rieck et al. [11] apply this algorithm to malware detection. Instead of fixing the number f of clusters, they use a threshold for the distances of values inside a cluster to the cluster head; hence the number of obtained clusters can differ. This functionality is also needed in our application. We however rely on the notion of similarity between two values (instead of distance) and provide a reformulation of the clustering algorithm here based on [10],[11]. The algorithm starts by assigning all values of the active domain to an initial cluster c₁, choosing an arbitrary element of it as h e a d₁ and then step by step choosing other head elements h e a d₂,…,h e a d _f that have lowest similarity to all other heads and moving other elements to the new clusters c₂,…,c _f ; an element is moved to a new cluster when it has higher similarity to the new head element than to the old head element. The algorithm continues finding new heads until a threshold α is reached; α limits the minimum similarity that elements inside a cluster may have to their cluster heads. Listing 2 shows a pseudocode for the clustering procedure.

Note that the clustering obtained by this heuristic is always complete: any value of π _A (F) is assigned to some cluster c _i . And we also have the property that clusters do not overlap: c _i ⋂c _j ≠θ for each i ≠ j.

When considering only a single relaxation attribute A, we obtain a fragmentation of the corresponding table: a set of fragments F _i - each corresponding to a cluster c _i . A relational algebra expression for each fragment can be stated as follows (using the selection operator σ and a disjunction of equality conditions on A for each value a contained in the cluster):

In the example query Q(x₁,x₂,x₃)=I l l(x₁,F l u) ∧ I l l(x₁,C o u g h) ∧ I n f o(x₁,x₂,x₃) the constant Cough is anti-instantiated. The fragment matching the Cough constant is the one containing respiratory diseases because we assume that it holds that s i m(F l u,C o u g h)>s i m(b r o k e n A r m,C o u g h). The second constant for the relaxation attribute in this query is Flu; however, Flu is a head element of the corresponding fragment and hence no similarities have to be computed. The anti-instantiated query is

The inequality condition on the new variables is necessary to only obtain answers where the two disease values found in the Respiratory fragment differ. A distributed join on x₁ has to be executed to combine the data from the Info table with the data from the Respiratory fragment; we will later on discuss how this overhead can be avoided by using derived fragmentation. Because the query is redirected to the fragment with highest similarity, in this case only the first informative answer (see Example 3) with the disease asthma I l l(2748,F l u) ∧ I l l(2748,A s t h m a) ∧ I n f o(2748,M a r y,` N e w S t r 3 , N e w t o w n `) is returned. In contrast, the answer for the disease brokenLeg is suppressed because it resides in the Fracture fragment.

The computation of distributed joins cannot be avoided if subqueries must be redirected to different server. We argue however, that with any other conventional data replication scheme (like [12],[14]), distributed joins have to be processed, too; while with our scheme we have added support for flexible query answering.

Consider the example query

The query has to be rewritten into the query

which has to be answered by both Fracture the and the Respiratory fragment. It may happen that the Respiratory, Fracture and Info tables all reside on different servers and so we would have to compute a three-way distributed join on x₁.

In our example we can join both the Treat as well as the Info table with the Ill table. Because we have two fragments of Ill, we obtain two derived fragments of Treat and Info as well: the first set of derived fragments is called Treat _resp and Info _resp based on a join on patient IDs occurring in the primary Respiratory fragment.

The second set of derived fragments is Treat _frac and Info _frac based on a join on patient IDs occurring in the primary Fracture fragment.

Note that non-redundancy of derived fragments is difficult to achieve (this is also discussed in [1]). We opt for having some redundancy in the derived fragments for sake of better data locality and hence better performance of query answering. That is why the information for patient Mary occurs in both derived fragments; the same applies to the treatment fragments.

We maintain separate lookup tables for each (primary and derived) fragmentation of each table. Hence, the sizes of the derived fragments are also computed and stored in the corresponding lookup table. These sizes of the derived fragments must be taken into account for the data replication procedure and are encoded in the BPPC as follows. The capacity W, the replication factor m, the primary fragments and their m - 1 copies as well as the conflict graph stay the same as before; the only input that changes is sizes w _i assigned to the fragments:

The Unified Medical Language System incorporates several taxonomies from the medical domain like the Systematized Nomenclature of Medicine-Clinical Terms (SNOMED CT), or Medical Subject Headings (MeSH). It unifies these taxonomies assigning Concept Unique Identifiers (CUI) to terms so that shared terms in the different taxonomies have the same identifier.

The Perl program UMLS::Similarity [15] offers an implementation of several standard similarity measures. They can be differentiated into measures based solely on path lengths in a taxonomy and measures taking the so-called information content [16] into account. The information content (IC) is computed from a pre-assigned estimated probability p(c) of each leave term in the taxonomy (assuming a parent-child or is-a relationship in the taxonomy); for inner nodes that subsume other terms, this probability must be larger than for any child node (for example, by summing over all child nodes) because this concept covers all its child concepts. The information content is then defined as the negative log likelihood: - logp(c). In this way, the higher a term is located in taxonomy, the more abstract the term it is, and the lower is its information content; where the unique root node of the taxonomy has IC 0 (or in other words, its probability is 1) - that is, no information content.

UMLS::Similarity offers implementations of the following measures based on path lengths:

UMLS::Similarity offers implementations of the following measures incorporating information content (IC):

We used the UMLS::Similarity web interface with the MeSH taxonomy to obtain the pair-wise similarity of the set of terms asthma, cough, influenza, tibial fracture and ulna fracture. Figure 1 shows how the terms are related by a is-a relationship in the MeSH taxonomy. Table 1 shows the similarity values obtained. Due to symmetry of the terms in the taxonomy (the path lengths and LCSs are mostly identical), the similarity values do not differ much in the two subsets asthma, cough and influenza, as opposed to tibial fracture and ulna fracture; the only difference is obtained with the two measures where the IC of the respective terms a, b are taken into account - namely jcn and lin.

The clustering heuristics has been implemented as a Java module that calls the UMLS::Similarity web interface. When using the clustering heuristics with the given similarities, regardless of which heads we choose, after two steps we obtain the two clusters {asthma, cough, influenza }, and {tibial fracture, ulna fracture }: let us assume, we choose asthma as h e a d₁, then we compute all similarities to asthma. The one with the lowest similarity is ulna fracture - which is taken to be h e a d₂. Because tibial fracture has lower similarity to ulna fracture than to asthma, it is assigned to c₂. For an appropriate threshold α (depending on the similarity measure chosen) the process could stop here. If instead we now continue the clustering, we would eventually obtain a total clustering consisting of only singleton sets: tibial fracture would become h e a d₃ (because it has minimal distance to h e a d₂); later on, cough would become h e a d₄ and influenza would be h e a d₅.

Choosing the path similarity and a threshold α=0.15 results in the two mentioned clusters. A fragmentation of the base table Ill can hence be obtained by computing the following materialized views in the Postgres database.

Sophisticated size estimations for these fragments might be possible as stated previously. We obtain the sizes of the fragment by counting the number of rows:

and

Next, we fill a lookup table containing information as.

To obtain the placement of the fragments to servers we model the corresponding BPPC and use the solver lp _solve [24]. lp_solve has a simple human-readable syntax and can be accessed by a Java program via the Java Native Interface (JNI). An example input for K=5 (maximum number of servers), W=5 (capacity per server), m=2 (replication factor) looks as shown in Listing 3.

For improved efficiency, the root table is currently stored as a hash map in the Java frontend (instead of stored in a separate database table). The hash map is keyed by the head element, because the heads are necessary for the query rewriting module.

The query rewriting procedure has to parse the SQL string inserted by the user. If a selection condition is given for the relaxation attribute, the root table is consulted to check for a matching head element. If none is found, again the UMLS::Similarity interface is consulted to obtain the similarities between head elements and the selection condition.

As a simple example considers the SQL query

When comparing bronchitis to the first head (that is, asthma), UMLS::Similarity gives the following similarity results: `The similarity of bronchitis (C0006277) and asthma (C0004096) using (jcn) is 1.0305, (cdist) is 0.3333, (lin) is 0.881, (wup) is 0.8, (path) is 0.3333, (res) is 3.5921, (lch) is 2.5903, (nam) is 0.1867’

Whereas comparing bronchitis to the second head (that is, tibial fracture), UMLS::Similarity gives the following similarity results: `The similarity of bronchitis (C0006277) and tibial fracture (C0040185) using (jcn) is 0.1308, (cdist) is 0.1429, (lin) is 0.2, (wup) is 0.5556, (path) is 0.1429, (res) is 0.9555, (lch) is 1.743, (nam) is 0.1483’

Hence, asthma is more similar to bronchitis in every measure. The SQL query is rewritten by retrieving the appropriate fragment name and redirected to the appropriate server identified from the root table:

That is, the entire fragment is returned as it is the one with the most relevant answers for the user.

In general, any flexible query answering approach incurs a certain performance overhead compared to exact query answering. In our case, the clustering and fragmentation have to be compute but also the query answering incurs some extra overhead due to the fact that the appropriate fragment has to be identified and multiple answers are returned whereas exact query answering would simply have returned an empty answer set. With our approach however we aim to reduce this overhead by locating all related answers in the same fragment; any other fragmentation approach would need to recombine related answers from different fragments. For a performance evaluation of our prototype we used a test dataset consisting of values taken from the list of Medical Subject Headings (MeSH) [25]. The similarity computation during the clustering constituted an extreme overhead. That is why we computed pairwise similarities for 300 sample headings and stored these similarity values in a separate table. With these 300 values we randomly filled the disease column of a test table. We varied the row count between 10 and 1000 rows. Another parameter to vary is the threshold α for the intra-cluster similarity: the maximum similarity that values in a cluster may have to their respective head. For a higher threshold, more clusters are computed (and hence more similarity computations are executed) than for a lower threshold. We tested similarity thresholds 0.1, 0.125, 0.3 and 0.5. We ran the clustering and fragmentation algorithm on a PC with 1.73 GHz and 4GB RAM. The observed runtimes and number of obtained fragments are reported in Table 2. For the lower threshold values 0.1 and 0.125 runtimes are in the range of some seconds up to around 17 minutes. For a row count of 1000 rows the higher similarity values lead to a high number of fragments and runtimes are hence prohibitively high. Due to the high amount of pairwise comparisons, obtaining the similarity values is still the bottleneck of the clustering procedure. In future work we will follow two ways of improving scalability of the clustering: optimizing the access to the similarity values and investigating implications of a parallel implementation of the clustering procedure.

The area of flexible query answering (sometimes also called cooperative query answering) has been studied extensively for single server systems. Some approaches have used taxonomies or ontologies for flexible query answering but did not consider their application for distributed storage of data: CoBase [26] used a type abstraction hierarchy to generalize values; Shin et al. [27] use some specific notion of metric distance in a knowledge abstraction hierarchy to identify semantically related answers; Halder and Cortesi [28] define a partial order between cooperative answers based on their abstract interpretation framework; Muslea [29] discusses the relaxation of queries in disjunctive normal form. Ontology-based query relaxation has also been studied for non-relational data (like XML data in [30]).

All these approaches address query relaxation at runtime while answering the query. This is usually prohibitively expensive. In contrast, our approach precomputes the clustering and fragmentation so that query answering does not incur a performance penalty.

There are some approaches for fine-grained fragmentation and replication on object/tuple level; however none of these approaches support the flexible query answering application aimed at in this paper. In contrast they are mostly workload-driven and try to optimize the locality of data that are covered in the same query. However, they only support exact query answering. In contrast to this, we do not consider workloads but a generic clustering approach that can work with arbitrary workloads providing the feature of flexible query answering by finding semantically related answers. While some approaches are adaptive to updates, no quality guarantee after an update is reported. We intend to extend our approach in the future by bringing robust optimization to the data replication area. Loukopoulos and Ahmad [6] describe data replication as an optimization problem; they focus on fine-grained geo-replication for individual objects. They include an assumed number of reads and writes for each site as well as communication costs between sites. They reduce their problem to the Knapsack problem. In particular, they devise an adaptive genetic algorithm that can reallocate data to different sites. We aim to follow a different path to support this adaptive behavior: the notion of robust optimization is briefly discussed in Section Discussion and conclusion.

Curino et al. [14] represent database tuples as nodes in a graph. They assume a given transaction workload and add hyperedges to the graph between those nodes that are accessed by the same transactions. By using a standard graph partitioning algorithm, they find a database fragmentation that minimizes the number of cut hyperedges. In a second phase, they use a machine learning classifier to derive a range-based fragmentation. Then they make an experimental comparison between the graph-based, the range-based, a hash-based fragmentation on tuple keys and full replication. Lastly, they also compare three different kinds of lookup tables to map tuple identifier to the corresponding fragment: indexes, bit arrays and Bloom filters. Similar to them, we apply lookup tables to locate the replicated data; however we apply this to larger fragments and not to individual tuples.

Quamar et al. [31] also model the fragmentation problem as minimizing cuts of hyperedges in a graph; for efficiency reasons, their algorithm works on a compressed representation of the hypergraph which results in groups of tuples. In particular, the authors criticize the fine-grained (tuple-wise) approach in [14] to be impractical for large number of tuples which is similar to our approach. The authors propose mechanisms to handle changes in the workload and compare their approach to random and tuple-level partitioning.

Tatarowicz et al. [12] assume three existing fragmentations: hash-based, range-based and lookup tables for individual keys and compare those in terms of communication cost and throughput. For an efficient management of lookup tables, they experimented with different compression techniques. In particular they argue that for hash-based partitioning, the query decomposition step is a bottleneck. While we apply the notion of lookup tables, too, the authors do not discuss how the fragments are obtained, whereas we propose a semantically guided fragmentation approach here.