Patent Description:
In a distributed relational data management system, data may be partitioned horizontally using a partitioning key to distribute rows of data into different partitions (e.g., physical or virtual storage devices, or portions thereof). In some embodiments, rows within a partition may be compressed in a row or columnar format forming files, to reduce the size of the data. This format allows for vectorized data processing and is more suitable for building a cloud based analytical engine. System performance tends to be better, in terms of compression and storage, when the file can be packed with a sufficient number of rows.

In a situation where there are not enough rows to form a pure file for one partition, rows belonging to different partitions can be combined to form an impure file. As multiple loads are received, impure files may be created with rows belonging to multiple partitions. Impure files may be described as belonging to a special partition called the "impure partition. " The impure partition includes cross partition data. In other words, impure files may be stored in or otherwise associated with the impure partition that contains only impure files. When enough rows are received to form a pure file (a file containing rows from a single partition), the pure file is stored in a "pure partition" (a partition containing only pure files). Although there may only be one impure partition, there may be a separate "pure partition" for each pure file. Having this one to one relationship enables increased partition elimination, allowing a scan process for data associated with a particular partition to scan a single pure partition and not scan other pure partitions. However, all of the files in the impure partition would also have to be scanned to find the files associated with the particular partition.

<NPL> discloses a method for file defragmention that re-orders blocks within segments.

Methods and systems are provided in a computing device for improved access to rows of data. Each data row is associated with a partition of a plurality of partitions. The data rows are distributed in one or more files. A file that includes data rows associated different partitions is an impure file. A clustering set is generated from a plurality of impure files. For example, a candidate file is selected from the plurality of impure files for inclusion in the clustering set based on file access activity metrics for the one or more impure files. One or more neighbor files are selected from the plurality of impure files for inclusion in the clustering set. Data rows of the impure files included in the clustering set are sorted according to their respective associated partitions. A set of disjoint partition range files are generated based on the sorted data rows of the impure files included in the clustering set. Each file of the set of disjoint partition range files is transferred to a respective target partition.

Further features and advantages of embodiments, as well as the structure and operation of various embodiments, are described in detail below with reference to the accompanying drawings. It is noted that the methods and systems are not limited to the specific embodiments described herein. Such embodiments are presented herein for illustrative purposes only. Additional embodiments will be apparent to persons skilled in the relevant art(s) based on the teachings contained herein.

The scope of the invention is defined by the independent claims. The
present specification and accompanying drawings disclose one or more embodiments that incorporate the features of the disclosed embodiments. The scope of the embodiments is not limited only to the aspects disclosed herein. The disclosed embodiments merely exemplify the intended scope, and modified versions of the disclosed embodiments are also encompassed. Embodiments are defined by the claims appended hereto.

Data that is stored in a distributed relational data management system may be partitioned horizontally using a partitioning key to distribute rows of data into different target partitions where they may be scanned in response to a query. Rows within a partition may be compressed in a row or columnar format to form files. This compressed format allows for vectorized data processing and is more suitable for building a cloud based analytical engine. Compression and storage performance depends upon having enough rows in a file to create a pure file stored in a pure partition. The pure partition may include rows of one file, and the pure file may have its rows stored in one pure partition. However, in instances where there are not enough rows to form a file in a pure partition (e.g., containing a predetermined number of rows), row belonging to multiple target partitions can be combined to form an impure file. For example, as multiple loads of data are received with an insufficient numbers of rows, an impure file is created having rows associated with multiple target partitions. One or more impure files may be stored together in a special partition referred to as an "impure partition. " In instances when enough rows are received to form a pure file, the pure file is stored in a pure partition.

As multiple sets of data loads are received in a data management system, files may accumulate in an impure partition. Multiple files stored in the impure partition can contain rows associated with the same target partition (e.g., a partition of interest during a user query scan). During a scan, in order to find all the rows associated with a particular partition of interest, in addition to reading the file in the (pure) partition of interest, all the files in the impure partition having rows associated with the partition of interest would also be read. Searching in multiple locations to find the queried data results in a significant overhead cost. For example, the more files stored in an impure partition, the more files that must be read to fetch rows during a scan, because the data rows for the target partition could potentially be spread across multiple files in the impure partition.

Methods and systems are provided to improve the clustering of files in an impure partition by grouping together rows in impure files that belong to the same target partition. As a result of grouping the rows associated with the same target partition (e.g., sorting based on an order key or a target partition identifier), when there are enough rows to form a file for a pure partition, a new pure file is created. Rows of the new pure file are all associated with the same target partition. The new pure file is then moved to the target partition (pure partition) and removed from the impure partition. The methods and systems provided herein include an online operation in cloud and non-cloud scenarios where the number of files in an impure partition are incrementally reduced to decrease (e.g., minimize) the impact on customer queries.

By implementing these methods online, with a goal of minimizing impact to overall system resources, the various embodiments of the solution become more challenging. For example, reducing the amount of data stored in an impure partition and moving the data to pure partitions for improved scanning efficiency, cannot be implemented effectively by simply reading all the rows of files stored in the impure partition into memory and sorting the rows by an order key (e.g., their target partition identifier (ID)) at the same time. In practice, this technique is not feasible because of resource constraints. For example, generally there is not enough memory available to sort all of the data of an impure partition without spilling. The present solution allows the system to perform operations on the impure files in a fault tolerant way without blocking incoming loads to the impure partition.

In some embodiments, the clustering improvement operations may be delegated to a background task, which may operate as part of a user's data management system (e.g., that executes user queries). The clustering improvement operations may share the same memory and compute resources with the user's system. Alternatively, or in addition, the clustering improvement operations may be delegated to a separate compute pool (e.g., in the cloud). These methods provide flexibility relative to resource constraints as well as the environment the clustering improvement system is running on. Given that the process might involve taking locks on certain files, which might interfere with the performance of user queries involving data present in these files, a policy may be set in place to cancel a clustering background task. The clustering background task may be rescheduled for execution after the user query has finished.

The disclosed clustering improvement methods include an incremental process that may be performed over multiple iterations to complete the clustering process. For example, instead of selecting all the files in the impure partition to sort, a subset of the files may be selected and sorted. By iteratively selecting and sorting subsets of files from an impure partition, a state of convergence may be reached where the rows of data in the impure files are clustered based on their order key (e.g., their target partition ID). During the process, a group of pure files (files where all rows belong to single target partition) are moved out of the impure partition to the corresponding target partition. At this point there may be some residue rows remaining in the impure partition. The decision on how many files to select at a time for a sorting operation may be based on system constraints and the current system load. The memory requirements for the clustering process may be proportional to the number of impure files to be sorted concurrently as well as the datatype of columns in the table schema. In instances where there are upper bounds on the file sizes, the memory requirements may be easily estimated.

For each iteration of impure file sorting, the clustering improvement system schedules background tasks in an order that is determined with a goal of improving user query performance (e.g., toward maximizing user query performance). The order of performing sorting tasks may be determined based on a static analysis of the clustering of data in the impure partition. Alternatively, or in addition, the order of the sorting tasks may be determined dynamically by prioritizing subsets of impure files accordingly to user workloads and which partitions or files are being queried more often. Methods and systems for prioritizing impure files for sorting tasks are described in more detail below. Note that the sorting tasks may be executed in parallel, as each sorting task may work on an independent dataset. The degree of parallelism may be decided for each iteration of sorting, based on the current system load and resource constraints. In instances where a particular sorting task fails, the system may re-schedule the task in a subsequent iteration, so as not to cause downtime to the user's system.

The methods and systems provided herein address how to select the impure files for each sorting iteration, and what may be a preferred way of selecting the impure files. A number of impure files to select for each sorting iteration may be selected in the cases that not all files of the impure partition may be selected at once. Moreover, methods are provided for determining a converged state (e.g., when the clustering process is complete or cannot make an additional improvement to the clustering of files in the impure partition).

Example embodiments are described as follows for incrementally improving clustering in an impure partition. An impure partition may include multiple impure files. Different rows in an impure file may be associated with different pure partitions, and rows associated with a particular pure partition may occur in multiple impure files. Impure files of an impure partition are sorted (e.g., clustered), and when enough rows are grouped in the impure partition to form a pure file, a new pure file containing the rows is created and moved to a target pure partition.

<FIG> and <FIG> are block diagrams of systems for incrementally improving clustering of cross partition data in a distributed data system, according to example embodiments. For example, referring to <FIG>, a system <NUM> for incrementally improving clustering of cross partition data in a distributed data system is shown, according to an embodiment. System <NUM> includes a computing system <NUM>, a storage device <NUM>, and a user query <NUM>. System <NUM> also includes a front end <NUM>, a compute node pool <NUM>, a distributed query processor <NUM>, compute nodes 112A, 112B, 112C, and 112D, and a cross partition data clusterer <NUM>. The group of compute nodes including 112A, 112B, 112C, and 112D may be referred to as compute nodes <NUM> or compute nodes 112A-112D. System <NUM> also includes a storage device <NUM>, a dataset <NUM>, a first pure partition 122A, a second pure partition 122B, a third pure partition 122C, a first impure file 124A, a second impure file 124B, a first clustered file 126A, a second clustered file 126B, a file <NUM>, and an impure partition <NUM>.

In general, computing system <NUM> is configured to store data in storage device <NUM>, and to respond to user queries by retrieving data from storage device <NUM>. Computing system <NUM> is also configured to process the data stored in storage device <NUM> and reorganize it for faster data retrieval and more efficient use of system <NUM> resources.

Computing system <NUM> may be a single computing device or may include multiple interconnected computing devices. For example, computing system <NUM> may include a computing device with front end <NUM> and distributed query processor <NUM>. Distributed query processor <NUM> may be communicatively coupled to compute nodes <NUM> via a network. Compute nodes <NUM> may each be a computing device that is operable to perform the embodiments described herein. In some embodiments, compute nodes <NUM> may function as a virtual machine managed by distributed query processor <NUM>.

As described above, computing system <NUM> and compute nodes <NUM> may include one or more computing devices. The one or more computing devices may include one or more computers, servers, mobile devices, etc. that are configured to communicate with storage device <NUM>. For example, computing system <NUM> may include one or more of a stationary computing device such as a desktop computer or personal computer, a super computer, a mobile computing device (e.g., a Microsoft® Surface® device, a personal digital assistant (PDA), a laptop computer, a notebook computer, a tablet computer such as an Apple iPad™, a netbook, etc.), a mobile phone (e.g., a cell phone, a smart phone such as a Microsoft Windows® phone, an Apple iPhone, a phone implementing the Google® Android™ operating system, etc.), a wearable computing device (e.g., a head-mounted device including smart glasses such as Google® Glass™, Oculus Rift® by Oculus VR, LLC, etc.), a gaming console/system (e.g., Microsoft Xbox®, Sony PlayStation®, Nintendo Wii® or Switch®, etc.), etc. In some embodiments, the functionality of computing system <NUM> may be included in a single device. The one or more computing devices of the computing system <NUM> are described in more detail with respect to <FIG> below.

Front end <NUM> is a storage interface that is configured to store data received from an external device in storage device <NUM>, and retrieve data from storage device <NUM> in response to a user query. When data is received from an external source, front end <NUM> may skip routing data to compute nodes <NUM> and directly write to storage device <NUM> (e.g., to impure or pure partitions). In instances when there are enough data rows to form a pure file, the front end <NUM> may compress the rows of data into a row or a columnar format to form a file and store the file in a pure partition, for example, as file <NUM> stored in third pure partition 122C. In instances where there are not enough data rows to compress into a pure file, front end <NUM> may be configured to label the rows of data or associate the rows of data with a partition identifier (ID) of a target pure partition, and combine the rows of data with data from one or more partitions to form an impure file. The impure file may then be compressed and stored in the impure partition <NUM>, such as first impure file 124A or second impure file 124B. Referring to <FIG>, first impure file 124A has data rows associated with multiple target pure partitions including first pure partition 122A and second pure partition 122B. Similarly, second impure file 124B has data rows associated with first pure partition 122A and second pure partition 122B. Although rows of impure files 124A and 124B are associated with first pure partition 122A and second pure partition 122B, impure files 124A and 124B are stored together in impure partition <NUM>. In other words, an impure file stored in impure partition <NUM> may be associated with two or more target pure partitions, and different impure files may each be associated with a different set of target pure partitions or the same set of target pure partitions.

Front end <NUM> may be a user interface that is further configured to receive data query signals such as user query <NUM>. User query <NUM> may include a user's request to retrieve data from storage device <NUM>. User query <NUM> may indicate a file to be retrieved from storage device <NUM>. Front end <NUM> is configured to forward information based on user query <NUM> to distributed query processor <NUM>.

In general, distributed query processing is the procedure of answering queries (e.g., performing read operations on large datasets) in a distributed environment where data may be managed at multiple sites in a computer network. Distributed query processor <NUM> is configured to assign or queue a task for one of compute nodes 112A - 112B (e.g., compute node 112A) from compute pool <NUM> to retrieve data rows requested in user query <NUM> from storage device <NUM>. Assigned compute node 112A is configured to determine a partition of interest associated with the requested file, and scan the partition of interest for the requested file. For example, the queried data may be in file <NUM> and the assigned compute node <NUM> may determine that the partition of interest is third pure partition 122C. The assigned compute node 112A may scan third pure partition 122C and retrieve file <NUM>. Additional rows of data associated with pure partition 122C may be stored in impure partition <NUM> that includes first impure file 124A and second impure file 124B. The assigned compute node 112A is configured to scan impure files 124A and 124B and retrieve any additional rows associated with the pure partition of interest (e.g., third partition 122C).

Storage device <NUM> may include one or more computer memory devices that may be communicatively coupled to computing system <NUM> and/or compute nodes <NUM>. For example, storage device <NUM> may include suitable logic, circuitry, and/or code to store and retrieve data for computing system <NUM>. Storage device <NUM> may include the dataset <NUM>, which is partitioned. For example, dataset <NUM> includes first pure partition 122A, second pure partition 122B, and third pure partition 122C. Dataset <NUM> may also include impure partition <NUM>. Each of the impure files 124A and 124B are stored in impure partition <NUM> and associated with first pure partition 122A and second pure partition 122B. For example, data rows or files within the impure files may be labeled with a partition ID corresponding to first pure partition 122A or second pure partition B. First pure partition 122A includes clustered file 126A, second pure partition 122B includes second clustered file 126B, and third pure partition 122C includes file <NUM>. First and second clustered files 126A and 126B are also pure files. Although storage device <NUM> is shown as a single device, storage device <NUM> may include a plurality of distributed storage devices (e.g., computer memory). Further aspects of storage device <NUM> are described with respect to <FIG>.

Computing system <NUM> also includes cross-partition data clusterer <NUM> that includes software executed by one or more of compute nodes <NUM>, such as compute node 112C. Cross-partition data clusterer <NUM> may be configured to sort data rows of impure files 124A and 124B stored in impure partition <NUM> according to an order key (e.g., a target partition ID) and store the sorted data rows in their respective target partitions (e.g., first pure partition 122A and second pure partition 122B).

Cross-partition data clusterer <NUM> may be configured in various ways, according to embodiments. For example, <FIG> is a block diagram of cross-partition data clusterer <NUM>, according to an example embodiment. As shown in <FIG>, cross-partition data clusterer <NUM> includes clustering set selector <NUM>, clusterer <NUM>, cluster transferor <NUM>, and file selection count determiner <NUM>. Cross-partition data clusterer <NUM> of <FIG> is described as follows, also with continued reference to system <NUM> of <FIG>.

Clustering set selector <NUM> of <FIG> is configured to select a subset of impure files (e.g., a clustering set) from impure partition <NUM> of <FIG> for sorting. As described above, memory constraints may prevent sorting all of the files in impure partition <NUM> at once. Instead of selecting all the impure files in impure partition <NUM> for sorting, only a subset of the impure files may be selected and sorted. File selection count determiner <NUM> may determine how many files to include in a sorting iteration. The number of impure files to sort may be determined based on system constraints and the current system load. For example, memory requirements for the sorting process may be proportional to the number of impure files selected to be sorted concurrently. In instances where there are upper bounds on file sizes, the memory requirements may be easily estimated.

Various methods may be used for selecting files of impure partition <NUM> for a clustering set. For example, for each iteration of sorting, clustering set selector <NUM> may select a subset of files (e.g., the clustering set) to be sorted. In some embodiments, clustering set selector <NUM> may score each of the impure files (e.g., based on quality or scan frequency), select a candidate impure file, and then select one or more neighbor impure files for the candidate file, to form the subset of impure files to be clustered. A candidate impure file and a neighbor impure file may be referred to as a candidate file and a neighbor file respectively. The file selection process may be divided into two phases. For example, a first phase may be a candidate file selection process where the candidate file is the first file identified for sorting (e.g., the choice of the candidate file may be determined based on input from customer workload data or static quality analyses data). A second phase may be a neighbor file selection process. For example, given the selected candidate file, clustering set selector <NUM> may determine which impure files to sort with the candidate file. The candidate file may then be sorted with the selected neighbor files. An example of candidate and neighbor file selection is described below.

As described above, system <NUM> may sort (or cluster) the data rows of selected impure files (from impure partition <NUM>) based on which target partition each data row belongs to, or is associated with. A key data structure may be created by clustering set selector <NUM> to keep track of which impure files of impure partition <NUM> have data rows associated each target partition. Table <NUM> below provides an example of impure files stored in an impure partition:.

Referring to Table <NUM>, the impure partition (e.g., impure partition <NUM>) includes six impure files with file IDs: <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>. The impure files are each associated with multiple target partitions of the four target partitions <NUM>-<NUM>. Table <NUM> below provides an example of a depth map which indicates the depth of each of the target partitions <NUM>-<NUM>:.

In Table <NUM>, the depth of a particular target partition is based on the number of impure files in impure partition <NUM> that have data rows associated with that particular target partition (i.e., the number of overlapping impure files associated with that particular target partition). This information may be useful in knowing which impure files have data rows for similar target partitions and should be sorted together.

A breadth map (or mathematical model) may be created to determine the "breadth" of an impure file. Breadth of an impure file may be defined as the sum of the depths of all target partitions having a depth greater than <NUM>, that the impure file is associated with. This information indicates the potential reduction of depth in the whole system (e.g., from impure partition <NUM> of dataset <NUM>) by clustering and removing data rows of an impure file (e.g., FILE11) from impure partition <NUM>. Target partitions with a depth equal to <NUM> are excluded from the breadth calculation as a depth equal to <NUM> means that the target partition does not have overlapping files. Referring to Tables <NUM> and <NUM>, the impure file with the greatest breadth is FILE <NUM>. FILE11 has a breadth of <NUM>, as it has data rows associated with all of the target partitions <NUM>-<NUM>. The sum of depths for target partitions <NUM> to <NUM> is: <NUM>+<NUM>+<NUM>+<NUM> = <NUM>. The breadth map may be used in "max breadth" type candidate and neighbor file selections as described below.

In some embodiments, the goal of candidate file selection is to select the first file to be included in the clustering set. A better technique for candidate file selection will yield faster convergence in the reduction of cross partition data, and/or fewer sorting iterations to reach convergence.

In some embodiments, a candidate impure file may be selected dynamically based on workload analysis (e.g., current work load, such a queue of one or more queries issued against rows in the file), or by static analysis of impure files and data distribution. An example of a dynamic candidate file selection process based on workload analysis may include monitoring various system metrics kept for the impure files. For example, various hotness (e.g., count of times a file is accessed) and cost metrics, per impure file, may be updated as the impure files are accessed for user queries (e.g., customer or user work load). For example, some of the metrics may indicate how many times an impure file is scanned for a user query. This information may be used to prioritize potential candidate impure files that are more important or relevant based on the current workload.

Various different processes may be utilized for selecting an impure file as a candidate file based on static analysis. In some embodiments, a "per partition" process may be used for candidate file selection. In this process, clustering set selector <NUM> may utilize a depth map to select the candidate file. For example, beginning with impure files associated with a first target partition having a depth greater than or equal to <NUM>, which may be a configurable threshold (e.g., target partition <NUM> shown in Table <NUM>), clustering set selector <NUM> may select a first impure file that has not been selected as a candidate file, and mark the selected impure file as discovered. This process may be repeated for the same target partition. If the depth for that target partition is greater than two and all the impure files associated with that target partition have been discovered, clustering set selector <NUM> may proceed to the next target partition in the depth map (e.g., target partition <NUM>). This process may continue until clustering set selector <NUM> has selected all the files in the depth map and there are no further candidate impure files.

In some embodiments, a process for selecting an impure file as a candidate file may be referred to as a "max depth process. " In this process, clustering set selector <NUM> may utilize a depth map to make the impure file selections. For example, clustering set selector <NUM> may select any impure file associated with the target partition that has the greatest depth in the depth map and mark the selected impure file as discovered. This process is repeated until the greatest depth of any target partition in the depth map is <NUM>, at which point there are no additional candidate files available. In this manner a file is selected for a target partition that has the maximum number of overlapping files in the impure partition.

In some embodiments, a process for selecting an impure file as a candidate file may be referred to as a "max breadth process. " The breadth of an impure file included in the depth map may be determined as the sum of the depths of all the target partitions of which rows of the impure file are associated, where only target partitions with depths having a configurable value greater than or equal to <NUM> are included in the summation. In the max breadth process, the impure file with the greatest breadth is selected as the candidate file and marked as discovered. This process may be repeated until the greatest breadth is zero, in which case there would be no further candidate files. In this manner, the file that is selected as the candidate file has the greatest number of overlaps with other impure files.

In some embodiments, the lowest depth in the depth map may be <NUM> because it is possible to have two impure files that overlap with respect to only one target partition and the overlap is at file boundaries. For example, the boundary of data rows for one impure file is also the boundary for data rows of the second impure file (e.g., File1 min partition == File2 max partition, or File1 max partition == File2 min partition). Therefore, in many cases, sorting these two impure files would always give same clustering result.

Clustering set selector <NUM> may also select n number of neighbor impure files for sorting with a given a candidate impure file. Various processes may be used to select the n neighbor files with a goal of reaching clustering convergence faster and/or with fewer clustering iterations.

In some embodiments, neighbor selection by clustering set selector <NUM> may be based on a process referred to as "per partition process. " In this process, using the depth map and starting from the candidate file's lowest partition ID to its highest partition ID (e.g., min partition ID to max partition ID), where the partition depth is greater than <NUM>, the first n impure files that have not yet been selected are selected and marked as selected. This is a simple approach for selecting the neighbor impure files for a candidate impure file.

In some embodiments, neighbor selection by clustering set selector <NUM> may be based on a process referred to as "max depth. " In this process, using the depth map and starting from the candidate file's lowest associated partition ID and going to its highest partition ID (e.g., min partition ID to max partition ID), the target partition with the greatest depth is selected. Then, the first n impure files associated with that target partition, that have not yet been selected, are selected as neighbor impure files, and marked as selected. In this manner, the impure files (neighbors) having the greatest number of overlapping files in the candidate file's partition range are selected as neighbors.

In some embodiments, neighbor selection by clustering set selector <NUM> may be based on a process referred to as "max breadth process. " In this process, from the impure files that overlap with the candidate impure file, n impure files with the greatest breadth values, and which have not yet been selected, are selected as neighbor files and marked as selected. In this manner, neighbor files that are selected are impure files that overlap with the candidate impure file and have the greatest number of overlaps with other impure files in the impure partition.

In some embodiments, neighbor selection by clustering set selector <NUM> may be based on a process referred to as "max overlap process. " In this process, from the impure files that overlap with the candidate file, the selected neighbor files are the n impure files with the highest number of target partitions in common with the candidate file (e.g., maximum overlapping partition range), and which have not yet been selected. The neighbor files are mark as selected. In this manner the impure files with the most overlapping partition range with the candidate file are selected as neighbor files.

In some embodiments, neighbor selection by clustering set selector <NUM> may be based on a process referred to as "max correlated histograms. " In this process, a histogram distribution is generated that indicates the number of rows in an impure file that map to each target partition. The number rows per target partition include minimum and maximum values. The correlation of the candidate's histogram to other overlapping impure files is determined and the impure files having the greatest correlation with the candidate's histogram are selected as the neighbor impure files.

In some embodiments, neighbor selection by clustering set selector <NUM> may be based on a process referred to as "max sum of overlapping partition rows. " In this process, for each of the impure files that overlap with the candidate file (e.g., on their overlapping partitions), the sum of the number of data rows for each impure file in the overlapping target partitions is determined. The impure files having the greatest number of data rows in the overlapping partitions are selected as neighbor impure files. In this manner, impure files that have the greatest number of data rows associated with the target partitions that overlap with the candidate file are selected as neighbors. The impure file histogram described above may be used by clustering set selector <NUM> to determine the number of data rows associated with each target partition in an impure file.

With reference to <FIG>, clusterer <NUM> is configured to sort data rows of the selected impure files based on order keys of the data rows (e.g., partition IDs corresponding to respective target partitions). Clusterer <NUM> may generate a set of files with disjoint partition ranges (e.g., each file of the set of files corresponds to a respective target partition). Cluster transferor <NUM> is configured to transfer each file of the generated files to their respective target partition, and remove the corresponding data rows from the impure partition (e.g., remove the impure files that were selected to be sorted, from the impure partition).

In some embodiments, the clustering set selector <NUM> selects a subset of impure files as a clustering set, and clusterer <NUM> sorts the data rows in the clustering set based on their partition IDs, and based on the results, generates files with disjoint target partition ranges (e.g., target partition ranges with an overlap of no more than <NUM>).

It may be assumed that a system with N impure files (files) with X rows each is given.

Partitioning function h:
row -> natural number <= K (partition), h(rj) = [<NUM>, K] for each row rj belonging to an impure file.

This function determines which target partition a given data row (rj) belongs to (or is associated with).

For each impure file, a histogram M returns a number of rows in the impure file mapping to a particular target partition or:
M(i) = count({ rj }), where i = [<NUM>,K] and { rj } is the collection of rows in the file for which all rj -> h(rj) = i.

A cluster/sort operation is defined as: given L number of files, where <NUM> < L << N, the data rows are re-arranged by creating L new files that obey the following conditions as true:.

The cost of the clustering operation is defined as <NUM> * L, as L files are read and L new files are written as a result.

By iteratively selecting, sorting, and transferring a subset of impure files from the impure partition to a pure partition, a state of convergence may be reached where there are no more files in the impure partition, or the rows of data in the impure partition are clustered based on their order key (e.g., their target partition ID). In this state, compute nodes <NUM> may scan for and retrieve data from data storage device <NUM> more quickly, and therefore, using fewer resources.

In embodiments, system <NUM> may operate in various ways to perform its functions. <FIG> shows a flowchart <NUM> for clustering cross partition data for improved user query operations, according to an example embodiment. In an embodiment, flowchart <NUM> may be performed by computing system <NUM> and storage device <NUM>. For the purposes of illustration, flowchart <NUM> of <FIG> is described with continued reference to <FIG> and <FIG>.

Flowchart <NUM> begins with step <NUM>. In step <NUM>, a clustering set is generated from a plurality of impure files. For example, impure partition <NUM> may include a plurality of impure files. First impure file 124A and second impure file 124B may comprise a subset of the impure files stored in impure partition <NUM>. Cross-partition data clusterer <NUM> may select a subset of files from impure partition <NUM> as a clustering set for a sorting process. For example, file selection count determiner <NUM> may determine the number of impure files to sort at one time. For example, the number of impure files may be based on a current system load metric, memory constraints, or a predicted number of sorting iterations needed to reach zero remaining impure files in the impure partition, or another convergence state. Clustering set selector <NUM> may select the determined number of impure files as a subset of the files stored in impure partition <NUM>. For example, clustering set selector <NUM> may identify first impure file 124A and second impure file 124B as the clustering set files.

In step <NUM>, data rows of the impure files included in the clustering set are sorted according to their respective associated partitions. Each data row (or file) of the impure files may be associated with a target pure partition. For example, a portion of the data rows of first impure file 124A and a portion of the data rows of impure file 124B are associated with first pure partition 122A. Also, a portion of the data rows of first impure file 124A and a portion of the data rows of impure file 124B are associated with second pure partition 122B. The data rows (or files) of first impure file 124A and second impure file 124B may be labeled with an order key, for example, a partition ID of the respective associated pure partition. Clusterer <NUM> may sort the data rows (or files) of the clustering set based on the order keys of the data rows (or files) in the cluster set.

In step <NUM>, a set of disjoint partition range files are generated based on the sorted data rows of the impure files included in the clustering set. For example, as a result of sorting the clustering set comprising first impure file 124A and second impure file 124B, the data rows of first impure file 124A and second impure file 124B may be ordered such that data rows corresponding to each particular target pure partition form a contiguous group of data rows (e.g., a set of disjoint partition range data rows). Clusterer <NUM> may generate a file for each group of data rows associated with the same target pure partition (e.g. a set of disjoint partition range files). For example, after clusterer <NUM> sorts the data rows of first impure file 124A and second impure file 124B, data rows associated with first pure partition 122A are grouped together and data rows associated with second pure partition 122B are grouped together. Clusterer <NUM> generates first cluster file 126A including data rows associated with first pure partition 122A and second cluster file 126B including data rows associated with second pure partition 122B. First cluster file 126A and second cluster file 126B are a set of disjoint partition range files.

In step <NUM>, each file of the set of disjoint partition range files are transferred to a respective target partition. For example, cluster transferor <NUM> transfers the first cluster file 126A to first pure partition 122A, and transfers second cluster file 126B to second pure partition 122B.

The process may be repeated until the number of data rows (or files) in the impure partition reaches zero or a convergent state. Thus response time to user queries may improve because compute nodes <NUM> have fewer partitions to scan when searching for a particular file or data rows. Moreover, by reducing query response time, compute resources are freed up for performing a greater number of user queries with the same amount of compute resources.

In embodiments, system <NUM> may operate in various ways to perform its functions. <FIG> is a flowchart for clustering cross partition data based on file access activity metrics for one or more impure files to improve user query operations, according to an example embodiment. In an embodiment, flowchart <NUM> may be performed by computing system <NUM> and storage device <NUM>. For the purposes of illustration, flowchart <NUM> of <FIG> is described with continued reference to <FIG> and <FIG>.

Flowchart <NUM> begins with step <NUM>, in step <NUM> a clustering set is generated from a plurality of impure files. For example, as described above, first impure file 124A and second impure file 124B may comprise a subset of the impure files stored in impure partition <NUM>. Distributed query processor <NUM> may assign or queue a task for a compute node of compute node pool <NUM> (e.g., compute node 112C) to select a subset of files from impure partition <NUM> as a clustering set for a sorting process. In response, cross-partition data clusterer <NUM> may select the clustering set. For example, the file selection count determiner <NUM> may determine the number of impure files to sort at one time. The number of impure files to sort may be based on a current system load metric, memory constraints, or a predicted number of sorting iterations needed to reach zero remaining impure files in the impure partition, or another convergence state. Clustering set selector <NUM> may select the determined number of impure files as a subset of the files stored in impure partition <NUM> as the clustering set. In some embodiments the clustering set includes a candidate file selected based on file access activity metrics for the one or more impure files and one or more neighbor files that are selected from the plurality of impure files for inclusion in the clustering set.

In step <NUM>, a candidate file is selected from the plurality of impure files for inclusion in the clustering set based on file access activity metrics for the one or more impure files. For example, the clustering set selector <NUM> may monitor various system metrics that are collected for the impure files. Hotness metrics (e.g., indicating a count of times a file is accessed) and resource cost metrics, per impure file, may be updated as the impure files are accessed for user queries (e.g., indicating customer or user work load on computing system <NUM> and storage device <NUM>). For example, some of the metrics may indicate how many times an impure file is scanned for a user query. This information may be used to prioritize potential candidate impure files. Clustering set selector <NUM> may select the candidate file based on current workload using the prioritized candidate impure files.

In step <NUM>, one or more neighbor files are selected from the plurality of impure files for inclusion in the clustering set. For example, in a neighbor file selection process, given the selected candidate file, clustering set selector <NUM> may determine which impure files to sort with the candidate file according to the number of impure files to sort determined by file selection count determiner <NUM>. Various processes may be used to select the neighbor impure files as described in more detail above with respect to <FIG> and <FIG>.

In step <NUM>, data rows of the impure files included in the clustering set are sorted according to their respective associated partitions. For example, each data row (or file) of the clustering set of impure files may be associated with a target pure partition. Compute node 112C may execute cross-partition data clusterer <NUM> and clusterer <NUM>. Clusterer <NUM> may sort the data rows (or files) of the clustering set based on respective order keys (e.g., respective target partition IDs) of the data rows (or files) in the cluster set.

In step <NUM>, a set of disjoint partition range files are generated based on the sorted data rows of the impure files included in the clustering set. For example, as a result of sorting the clustering set the data rows of the candidate impure file and neighbor impure files may be ordered such that data rows corresponding to each particular target pure partition form a contiguous group of data rows (e.g., a set of disjoint partition range data rows). Clusterer <NUM> may generate a file for each group of data rows associated with the same target pure partition (e.g. a set of disjoint partition range files) and store the corresponding data rows in the generated files.

In step <NUM>, each file of the set of disjoint partition range files are transferred to a respective target partition. For example, each of the generated files is a pure file including only data rows associated with a respective target partition and are stored in the respective target partition, which is a pure partition.

The process of steps <NUM> through <NUM> may be repeated until the number of data rows (or files) in the impure partition reaches zero or a convergent state. In this manner, response time to user queries may improve because compute nodes <NUM> have fewer files to scan when searching for particular data belonging to a partition or partition ID. Moreover, by reducing query response time, compute resources are freed up for performing a greater number of user queries with the same amount of compute resources.

In the above example embodiments, compute node 112C and cross-partition data clusterer <NUM> perform all of the steps <NUM>-<NUM> and <NUM>-<NUM>. However, in other embodiments, the steps or portions of the steps performed by cross-partition data clusterer <NUM> may be performed by cross-partition data clusterers <NUM> of multiple compute nodes in compute node pool <NUM>. For example, one compute node of compute node pool <NUM> may execute functions of file selection count determiner <NUM> and clustering set selector <NUM> while another compute node of compute node pool <NUM> executes functions of clusterer <NUM> and cluster transferor <NUM>. However, the disclosure is not limited in this regard and execution of the steps and/or portions of steps <NUM>-<NUM>, steps <NUM>-<NUM>, and/or other steps, methods, or systems described herein, may be distributed in any suitable manner among the compute nodes of compute node pool <NUM>. Methods and systems for distributing tasks for incrementally improving clustering of cross-partition data in a distributed data system are described in more detail below.

A system such as system <NUM> for improving cross partition clustering may be implemented according to various architectures to perform its functions. <FIG> is a block diagram of an example architecture for executing user workloads and incremental cluster improvement workloads on shared compute nodes using background threads, according to an example embodiment. Referring to <FIG>, there is shown a system <NUM> including compute nodes 410A-410C, clustering task queue <NUM>, and impure partition <NUM>. System <NUM> also includes cross-partition data clusterers 114A - 114C, and background threads 420A-420C.

System <NUM> is an example architecture that may be utilized to implement system <NUM>. Compute nodes 410A-410C may be similar or substantially the same as compute nodes 112A-112D. Cross-partition data clusterer 114A-114C may be similar or substantially the same as cross-partition data clusterer <NUM>. Impure partition <NUM> may be similar or substantially the same as impure partition <NUM>. Background threads 420A-420C may execute tasks for incremental cluster improvement in the storage device <NUM> depending on the implemented system architecture.

In a data warehouse, such as in system <NUM>, where large amounts of data are partitioned, several compute nodes may scan partitioned data. For example, in response to a user query, each compute node 410A-410C may scan a single pure partition or set of pure partitions and may also query the impure partition <NUM> as it may include data rows associated with a pure partition of interest in the query. As described above, one goal of the disclosed embodiments is to improve clustering in impure partition <NUM>, which may eventually lead to creating a pure file. The pure file may be moved to a respective target pure partition. By doing so, compute nodes 410A-410C no longer need to query impure partition <NUM> for responding to a user query, because, the files have been transferred to the target pure partition, thereby improving overall query performance.

By design of the system architecture, the impure file selection processes described above may be decoupled from the clustering processes (e.g., sorting processes). For example, in system <NUM> a compute node 410A, which may be configured to execute file selection processes, may be decoupled from other compute nodes 410B and 410C that may be configured to execute the clustering processes. In system <NUM>, there are multiple compute nodes 410A-410C that may access impure partition <NUM>. Compute node 410A may be configured to execute an impure file selection process based on impure files stored in impure partition <NUM>, and may also be configured to schedule a clustering task in clustering task queue <NUM> for the selected impure files. For example, clustering task queue <NUM> may be configured as a global queue that is accessible by all of compute nodes 410A-410C. Compute node 410B or compute node 410C may be configured to retrieve the clustering task from clustering task queue <NUM> and execute a sorting process (e.g., a sorting iteration) on the files selected by compute node 410A.

By executing clustering tasks across multiple compute nodes, clustering (e.g., sorting) may be performed faster while the cost of clustering is shared across multiple compute nodes (e.g., compute nodes 410A-410C). The cost per compute node can vary depending on how much of the clustering task load each compute node is configured to perform at a time. As shown in <FIG>, each of compute nodes 410A, 410B, and 410C has a respective cross partition data clusterer 114A, 114B, or 114C, and has a configurable number (n) of background threads per compute node (e.g., 420A, 420B, or 420C). Background threads 420A, 420B, and 420C may be configured to execute selection and/or clustering tasks of the respective cross-partition data clusterers 114A, 114B, and 114C as described above with respect to system <NUM>. For example, background thread 420B may be configured to retrieve a clustering task from cluster task queue <NUM>, execute sorting of the selected impure files from impure partition <NUM>, generate a set of disjoint partition range files as a result of the sorting, transfer each file of the set of disjoint partition range files to a respective target partition.

The architecture shown in <FIG> may be configured to prevent a background thread, which is configured to perform clustering of impure files, from consuming too large a portion of a compute node's resources and thereby reduce the availability of resources for user queries (e.g., slowing down query speed). As the number of files being clustered at one time increases, the amount of resources needed to execute the cluster process also increases. Therefore, the number of files to be clustered in a clustering cycle is automatically and dynamically configurable by the file section count determiner <NUM> before each file selection iteration to control the percentage of resources allocated to clustering processes versus the user query work load. Moreover, in some embodiments, a compute node <NUM> that is too busy with user queries may decide not to retrieve any new clustering tasks from the clustering task queue <NUM>, until it spare resources are available.

In one embodiment, background thread 420A on compute node 410A may be configured to execute a file selection process and queue a clustering task for the selected impure files. The other background threads 420B and 420C may be configured to retrieve the clustering task from the queue. In other embodiments, multiple background threads, e.g., background threads 420A and 420B may be configured to execute file selection processes, and/or multiple background threads may be configured for executing clustering processes, for example, depending on how many impure files are stored in the impure partition <NUM>. An advantage of the architecture shown in system <NUM> is that spare resources may be opportunistically utilized for clustering impure files, on compute nodes 410A-410C that also handle user workloads. Resource consumption for clustering processes on each compute node may be dynamically increased or decreased depending on the spare resources available, which aren't being used for user workload. When the compute nodes are idle, all of the available resources may be used for clustering. In this manner, the speed of the clustering processes may be significantly increased and will in turn benefit the customer workloads in the system.

As described above, a system such as system <NUM> for improving cross partition clustering may be implemented according to various architectures to perform its functions.

<FIG> is a block diagram of an example architecture for executing user workloads and incremental cluster improvement workloads in separate compute node pools, according to an example embodiment.

Referring to <FIG>, there is shown a system <NUM> including compute nodes 510A-510D, clustering task queue <NUM>, and impure partition <NUM>. System <NUM> also includes cross-partition data clusterers 114A - 114B and background threads 520A-520D.

System <NUM> is an example architecture that may be utilized to implement system <NUM>. Compute nodes 510A-510D may be similar or substantially the same as compute nodes 112A-112D. Cross-partition data clusterer 114A-114B may be similar or substantially the same as cross-partition data clusterer <NUM>. Impure partition <NUM> may be similar or substantially the same as impure partition <NUM>. Background threads 520A-520B may execute processes of cross-partition data clusterer 114A and background threads 520C-520D may execute processes of cross-partition data clusterer 114B. For example, background threads 520A-520D may execute tasks for incremental cluster improvement in storage device <NUM> depending on the implemented system architecture.

In the architecture of system <NUM>, the clustering system is separated from the user query system by having separate compute pools for user query workloads and impure file clustering workloads. With this type of architecture, the clustering compute pool may be scaled independently from the user query compute pool. Referring to <FIG>, a user query compute node pool may include compute nodes 510A and 510B. A clustering compute node pool may include compute nodes 510C and 510D. Background threads 520A-520D of the clustering compute node pool may access clustering task queue <NUM> to retrieve clustering tasks to execute. In some embodiments, background threads 520A-520B of compute node 510C may be configured to execute selection of a subset of impure files from impure partition <NUM>, and background threads 520C-520D may be configured to execute sorting of the selected impure files for transferring of data rows of the impure files from impure partition <NUM> to respective target partitions. This system <NUM> architecture advantageously provides a dedicated compute node pool for clustering processes, in which the entire compute resources of the dedicated compute pool may be used for the clustering process and may be scaled independently of the compute node pool dedicated to user query workloads.

Example capabilities of the embodiments of systems <NUM>, <NUM> and <NUM> include:.

Embodiments described herein may be implemented in hardware, or hardware combined with software and/or firmware. For example, embodiments described herein may be implemented as computer program code/instructions configured to be executed in one or more processors and stored in a computer readable storage medium. Alternatively, embodiments described herein may be implemented as hardware logic/electrical circuitry.

As noted herein, the embodiments described, including but not limited to, system <NUM> of <FIG> and <FIG>, system <NUM> of <FIG>, system <NUM> of <FIG>, system <NUM> of <FIG>, and system <NUM> of <FIG>, along with any components and/or subcomponents thereof, as well any operations and portions of flowcharts/flow diagrams described herein and/or further examples described herein, may be implemented in hardware, or hardware with any combination of software and/or firmware, including being implemented as computer program code configured to be executed in one or more processors and stored in a computer readable storage medium, or being implemented as hardware logic/electrical circuitry, such as being implemented together in a system-on-chip (SoC), a field programmable gate array (FPGA), an application specific integrated circuit (ASIC), a trusted platform module (TPM), and/or the like. A SoC may include an integrated circuit chip that includes one or more of a processor (e.g., a microcontroller, microprocessor, digital signal processor (DSP), etc.), memory, one or more communication interfaces, and/or further circuits and/or embedded firmware to perform its functions.

Embodiments described herein may be implemented in one or more computing devices similar to a mobile system and/or a computing device in stationary or mobile computer embodiments, including one or more features of mobile systems and/or computing devices described herein, as well as alternative features. The descriptions of computing devices provided herein are provided for purposes of illustration, and are not intended to be limiting. Embodiments may be implemented in further types of computer systems, as would be known to persons skilled in the relevant art(s).

<FIG> is a block diagram of an example processor-based computer system that may be used to implement various embodiments. Computing device <NUM>, storage device <NUM>, and compute nodes 112A-112D, 410A-410C, and 510A-510D may each include any type of computing device, mobile or stationary, such as a desktop computer, a server, a video game console, etc. For example, any of computing device <NUM>, storage device <NUM>, and compute nodes 112A-112D, 410A-410C, and 510A-510D may be any type of mobile computing device (e.g., a Microsoft® Surface® device, a personal digital assistant (PDA), a laptop computer, a notebook computer, a tablet computer such as an Apple iPad™, a netbook, etc.), a mobile phone (e.g., a cell phone, a smart phone such as a Microsoft Windows® phone, an Apple iPhone, a phone implementing the Google® Android™ operating system, etc.), a wearable computing device (e.g., a head-mounted device including smart glasses such as Google® Glass™, Oculus Rift® by Oculus VR, LLC, etc.), a stationary computing device such as a desktop computer or PC (personal computer), a gaming console/system (e.g., Microsoft Xbox®, Sony PlayStation®, Nintendo Wii® or Switch®, etc.), etc..

<FIG> depicts an exemplary implementation of a computing device <NUM> in which embodiments may be implemented. For example, computing device <NUM>, storage device <NUM> and compute nodes 112A-112D, 410A-410C, and 510A-510D may each be implemented in one or more computing devices similar to computing device <NUM> in stationary or mobile computer embodiments, including one or more features of computing device <NUM> and/or alternative features. The description of computing device <NUM> provided herein is provided for purposes of illustration, and is not intended to be limiting. Embodiments may be implemented in further types of computer systems, as would be known to persons skilled in the relevant art(s).

A number of program modules may be stored on the hard disk, magnetic disk, optical disk, ROM, or RAM. These programs include operating system <NUM>, one or more application programs <NUM>, other programs <NUM>, and program data <NUM>. Application programs <NUM> or other programs <NUM> may include, for example, computer program logic (e.g., computer program code or instructions) for implementing computing device <NUM>, storage device <NUM>, a front end <NUM>, compute node pool <NUM>, distributed query processor <NUM>, compute nodes 112A, 112B, 112C, and 112D, cross partition data clusterer <NUM>, clustering set selector <NUM>, clusterer <NUM>, cluster transferor <NUM>, file selection count determiner <NUM>, compute node 410A, compute node 410B, compute node 410C, cross-partition data clusterer 114A, cross-partition data clusterer 114B, cross-partition data clusterer 114C, background thread 420A, background thread 420B, background thread 420C, clustering task queue <NUM>, impure partition <NUM>, compute node 510A, compute node 510B, compute node 510C, compute node 510D, cross-partition data clusterer 114A, cross-partition data clusterer 114B, background thread 520A, background thread 520B, background thread 520C, background thread 520D, flowchart <NUM>, flowchart <NUM>, and/or further embodiments described herein. The program data <NUM> may include dataset <NUM>, first pure partition 122A, second pure partition 122B, third pure partition 122C, first impure file 124A, second impure file 124B, first clustered file 126A, second clustered file 126B, file <NUM>, impure partition <NUM>, impure partition <NUM>, clustering task queue <NUM>, and/or further embodiments described herein.

A user may enter commands and information into computing device <NUM> through input devices such as keyboard <NUM> and pointing device <NUM>.

In an embodiment, a system for improved access to rows of data, where each data row is associated with a partition of a plurality of partitions, the data rows are distributed in one or more files, wherein a file including data rows associated with different partitions of the plurality of partitions is an impure file, the system comprises: one or more processors and one or more memory devices that store program code to be executed by the one or more processors. The program code comprises a clustering set selector that is configured to generate a clustering set from a plurality of impure files. A clusterer is configured to: sort data rows of the impure files included in the clustering set according to their respective associated partitions, and generate a set of disjoint partition range files based on the sorted data rows of the impure files included in the clustering set. A cluster transferor is configured to: transfer each file of the set of disjoint partition range files to a respective target partition.

In an embodiment, the clustering set selector is further configured to: select a candidate file from the plurality of impure files for inclusion in the clustering set, and select one or more neighbor files from the plurality of impure files for inclusion in the clustering set. The candidate file is selected independent of the selection of the one or more neighbor files.

In an embodiment, the clustering set selector is further configured to select the candidate file based on: file access activity metrics for the one or more impure files, analysis of a number of partitions associated with each of the plurality of impure files, or analysis of a number of impure files associated with a partition.

In an embodiment, the clustering set selector, the clusterer, and the cluster transferor are configured to iterate until a number of impure files reaches zero or another convergence state is reached.

In an embodiment, the clustering set selector is executed independently of the clusterer.

In an embodiment, the clusterer is executed: in one or more background threads executed on one or more compute nodes of the one or more processors, wherein the one or more compute nodes are also configured to execute user queries; or in a dedicated compute node pool of the one or more processors, wherein the dedicated compute node pool is configured to execute the cluster set selector, the clusterer, and the cluster transferor, and other compute nodes execute user queries.

In an embodiment, execution of a clustering task by the clusterer is cancelled and rescheduled in response to interference, by the clusterer, of performance of user queries involving the impure files present in the clustering set.

In an embodiment, the system further comprises: a file selection count determiner configured to determine a number of the plurality of impure files to include in the clustering set based on at least one of: a system load metric; memory constraints, or a predicted number of sorting iterations needed to reach zero remaining impure files or another convergence state.

In an embodiment, a method in a computing device for improved access to rows of data, where each data row is associated with a partition of a plurality of partitions, the data rows are distributed in one or more files, wherein a file including data rows associated with different partitions of the plurality of partitions is an impure file, the method comprises: generating a clustering set from a plurality of impure files; sorting data rows of the impure files included in the clustering set according to their respective associated partitions; generating a set of disjoint partition range files based on the sorted data rows of the impure files included in the clustering set; and transferring each file of the set of disjoint partition range files to a respective target partition.

In an embodiment, said generating a clustering set from a plurality of impure files comprises: selecting a candidate file from the plurality of impure files for inclusion in the clustering set, and selecting one or more neighbor files from the plurality of impure files for inclusion in the clustering set; wherein the candidate file is selected independent of the selection of the one or more neighbor files.

In an embodiment, the candidate file is selected based on: file access activity metrics for the one or more impure files, analysis of a number of partitions associated with each of the plurality of impure files, or analysis of a number of impure files associated with a partition.

In an embodiment, steps of said generating, sorting, generating, and transferring are iterated until a number of impure files reaches zero or another convergence state is reached.

In an embodiment, said generating a clustering set from a plurality of impure files is executed independent of execution of said sorting data rows of the impure files included in the clustering set according to their respective associated partitions.

In an embodiment, said sorting data rows of the impure files included in the clustering set according to their respective associated partitions is executed in: one or more background threads executed on one or more compute nodes of the one or more processors, wherein the one or more compute nodes are also configured to execute user queries; or a dedicated compute node pool of the one or more processors, wherein the dedicated compute node pool is configured to execute the cluster set selector, the clusterer, and the cluster transferor, and other compute nodes execute user queries.

In an embodiment, execution of the sorting of data rows of the impure files is cancelled and rescheduled in response to interference of user queries involving the impure files present in the clustering set by the sorting of the impure files.

In an embodiment, the method further comprises: determining how a number of the plurality of impure files to include in the clustering set based on at least one of: a system load metric; memory constraints, or a predicted number of sorting iterations needed to reach zero remaining impure files or another convergence state.

In an embodiment, a method in a computing device for improved access to rows of data, where each data row is associated with a partition of a plurality of partitions, the data rows are distributed in one or more files, wherein a file including data rows associated with different partitions of the plurality of partitions is an impure file, the method comprises: generating a clustering set from a plurality of impure files including: selecting a candidate file from the plurality of impure files for inclusion in the clustering set based on file access activity metrics for the one or more impure files, and selecting one or more neighbor files from the plurality of impure files for inclusion in the clustering set; sorting data rows of the impure files included in the clustering set according to their respective associated partitions; generating a set of disjoint partition range files based on the sorted data rows of the impure files included in the clustering set; and transferring each file of the set of disjoint partition range files to a respective target partition.

In an embodiment, the method further comprises: determining how many of the plurality of impure files to include in the clustering set based on at least one of: a system load metric; memory constraints, or a predicted number of sorting iterations needed to reach zero remaining impure files or another convergence state.

Claim 1:
A system for improved access to rows of data, the data rows being distributed into a plurality of target partitions (122A-C), before being compressed in one or more files (124A-B, 126A-B, <NUM>), wherein a file including data rows belonging to different target partitions of the plurality of target partitions is an impure file (124A-B), wherein a target partition is a partition of interest during a user query scan, the system comprising:
one or more processors (<NUM>);
one or more memory devices (<NUM>) that store program code to be executed by the one or more processors, the program code comprising:
clustering set selector (<NUM>) configured to:
generate a clustering set from a plurality of impure files;
a clusterer (<NUM>) configured to:
sort data rows of the impure files included in the clustering set according to their respective target partitions, and
generate a set of disjoint partition range files each comprising only data rows belonging to a given target partition based on the sorted data rows of the impure files included in the clustering set; and
a cluster transferor (<NUM>) configured to:
transfer each file of the set of disjoint partition range files to the respective target partition.