Patent Description:
The present disclosure relates systems, methods, and devices for a multi-cluster warehouse.

Databases are widely used for data storage and access in computing applications. Databases may include one or more tables that include or reference data that can be read, modified, or deleted using queries. Databases can store small or extremely large sets of data within one or more tables. This data can be accessed by various users in an organization or even be used to service public users, such as via a website or an application program interface (API). Both computing and storage resources, as well as their underlying architecture, can play a large role in achieving desirable database performance. <CIT> discloses an example resource provisioning systems and methods are described. In one implementation, an execution platform accesses multiple remote storage devices. The execution platform includes multiple virtual warehouses, each of which includes a cache to store data retrieved from the remote storage devices and a processor that is independent of the remote storage devices. A resource manager is coupled to the execution platform and monitors received data processing requests and resource utilization. The resource manager also determines whether additional virtual warehouses are needed based on the data processing requests and the resource utilization. If additional virtual warehouses are needed, the resource manager provisions a new virtual warehouse. XP2793632 discloses: "One of the oldest challenges in data warehousing is getting data into the data warehouse without getting in the way of other queries and workloads. In this Snowflake Chaiienge, we're going to demonstrate how Snowflake makes that possible with concurrent load and query. "
XP55416980 discloses: "We live in the golden age of distributed computing. Public cloud plat. fUrms now offer virtually unlimited compute and storage resources on demand. At, the same time, the Software-as-a-Service (Sans. ) model brings enterprise-class systems to users who previously could not afford such systems due to their cost and complexity. Alas, traditional data warehousing systems are struggling to fit into this new environment. For one thing, they have been designed for fixed resources and are thus unable to leverage the cloud's elasticitv. For another thing, their dependence on complex ETL pipelines and physical tuning is at odds with the flexibility and freshness requirements of the cloud's new types of semi-structured data and rapidly evolving workloads.

We decided a fundamental redesign was in order. Our mission was to build an enterprise-ready data warehousing solution for the cloud- The result is the Snowflake Elastic Data Warehouse, or "Snowflake" for short. Snowflake is a multi-tenant, transactional, secure, highly scalable and elastic system with full SQL support and built-in extensions for semi-structured and schema-less data. The system is offered as a pay-as-you-go service in the Amazon cloud. Users upload their data to the cloud and can immediately manage and query it using familiar tools and interfaces. Implementation began in late <NUM> and Snowflake has been generally available since June <NUM>. Today, Snowflake is used in production by a growing number of small and large organizalions alike. The system runs several million queries per day over multiple petabytes of data.

In this paper, we describe the design of Snowflake and its novel multi-cluster, shared-data architecture. The paper highlights some of the key features of Snowflake: extreme elasticity and availability, semi-structured and schema-less data, time travel, and end-to-end security. It, concludes with lessons learned and an outlook on ongoing work.

Non-limiting and non-exhaustive embodiments of the present disclosure are described with reference to the following figures, wherein like reference numerals refer to like parts throughout the various figures unless otherwise specified.

The present disclosure is directed to system, methods, and devices for providing and managing multi-cluster warehouses. A warehouse is several servers that are connected and collaborate in handling analytical queries. In some warehouses, compute and storage resources are connected and allocated together. In at least some embodiments disclosed herein, compute resources are independently allocated and scalable separate from storage resources. In some cases, a warehouse includes one or more clusters and/or a cluster of one or more server nodes that can work together to provide services. Applicants have developed, and herein present, architectures, methods, algorithms, and systems for multi-cluster warehouses.

According to one embodiment, a method for a multi-cluster warehouse includes allocating a plurality of compute clusters as part of a virtual warehouse. The compute clusters may be used to access and perform queries against one or more databases in one or more cloud storage resources. The method includes providing queries for the virtual warehouse to each of the plurality of compute clusters. For example, each of the plurality of compute clusters of the virtual warehouse may receive a plurality of queries so that the computing load is spread across the different clusters. The method may also include dynamically adding compute clusters to and removing compute clusters from the virtual warehouse as needed based on a workload of the plurality of compute clusters.

A multi-cluster warehouse can provide significant improvements in concurrency as well as availability. For example, a warehouse generally includes only one single cluster whose size is the size of the warehouse. For example, a large warehouse may include a single cluster of eight server nodes. A multi-cluster warehouse may allow for creation of a single warehouse with multiple clusters. Each cluster within the warehouse may include eight server nodes. Thus, the multi-cluster warehouse may support three times the level of concurrency provided by a single cluster warehouse of the same size. This architecture can allow for a high level of concurrency against a single warehouse while also allowing for scaling of computing resources, as will be discussed further herein.

Improved availability can also be achieved in a multi-cluster warehouse by placing different clusters in different availability zones. For example, multi-cluster warehouses will provide improved fault resilience since each warehouse cluster could be allocated in a different availability zone of a cloud provider (such as within different Amazon ® availability zones. Hence, a multi-cluster warehouse would be highly available compare to a single-cluster warehouse. Furthermore, queries can be routed to an optimal cluster where relevant data segments are already in memory or local disk-based storage (e.g., in a cache). For example, a method for a multi-cluster warehouse may include routing queries based on a session from which the query originated. By providing queries from a same session to a same cluster, a likelihood is increased that the data needed for a query is already in memory and may eliminate a need to retrieve that data from a cloud storage resource. With improved concurrency and availability, users may experience improved response times and availability that would be difficult or impossible to achieve in other traditional single-cluster database architectures.

In addition to improved availability and concurrency, significant variation in automatic scaling of compute resources is possible. For example, at least some embodiments provide separate allocation of compute resources from cloud storage. Thus, a multi-cluster warehouse can be scaled up or down in the number of compute clusters to accommodate wide swings in workload while still querying against the data that has not changed or is changing very slowly compared to the query workload.

Automatically resuming or starting a new or suspended cluster may be performed when the warehouse cannot handle the workload and would have to queue queries (or queue queries longer than an accepted length of time). Queries can get queued because the total resource consumption on the cluster has exceeded a threshold. For example, the resource consumption may include parameters for memory load as well as computing or processing load. In one embodiment, a parameter controls for how long a query may be queued before a new cluster should be resumed or provisioned. As soon as the new cluster has resumed, queries can be scheduled to execute on the new cluster. This applies to new queries as well as already queued queries.

In one embodiment, a method for a multi-cluster warehouse may include dynamically adding compute clusters to the virtual warehouse based on the workload. The method may include determining whether a query can be processed while meeting a performance metric for the query. If the query in combination with a current workload does not allow one or more currently allocated compute clusters to meet the performance metric, the method may include triggering startup of a new compute cluster. In some embodiments, a new cluster can be allocated quickly enough to ensure that not a single query experiences less than the required performance metric.

Auto-suspending or decommissioning an active cluster of a multi-cluster warehouse may be performed when the resource consumption of the workload is low enough that suspending that cluster would not have caused any query executed in the past N minutes to be queued (or queued longer than a threshold time). The queueing of a query or a queuing time for a query is just one example of a performance metric that may be used. In one embodiment, a method for a multi-cluster warehouse may include removing compute clusters based on the workload. The method may include determining whether a current workload is serviceable by one fewer than the plurality of compute clusters while meeting a performance metric. The method may include decommissioning (or suspending) at least one compute cluster of the plurality of compute clusters in response to determining that the workload is serviceable by one fewer than the plurality of compute clusters.

According to one embodiment, automatic provisioning or removal of clusters, as well as routing queries to different clusters within a warehouse, may be used as part of a powerful and flexible multi-cluster warehouse as a service.

A detailed description of systems and methods consistent with embodiments of the present disclosure is provided below.

Turning to the figures, <FIG> is a block diagram illustrating a processing platform <NUM> for providing and/or managing a multi-cluster warehouse, according to one embodiment. The processing platform <NUM> includes a resource manager <NUM> that is accessible by multiple users <NUM>, <NUM>, and <NUM>. The resource manager <NUM> may also be referred to herein as a database service manager. In some implementations, resource manager <NUM> can support any number of users desiring access to data or services of the processing platform <NUM>. Users <NUM>-<NUM> may include, for example, end users providing data storage and retrieval queries and requests, system administrators managing the systems and methods described herein, software applications that interact with a database, and other components/devices that interact with resource manager <NUM>.

The resource manager <NUM> may provide various services and functions that support the operation of the systems and components within the processing platform <NUM>. Resource manager <NUM> has access to stored metadata <NUM> associated with the data stored throughout data processing platform <NUM>. The resource manager <NUM> may use the metadata <NUM> for optimizing user queries. In some embodiments, metadata <NUM> includes a summary of data stored in remote data storage systems as well as data available from a local cache (e.g., a cache within one or more of the clusters of the execution platform <NUM>). Additionally, metadata <NUM> may include information regarding how data is organized in the remote data storage systems and the local caches. Metadata <NUM> allows systems and services to determine whether a piece of data needs to be processed without loading or accessing the actual data from a storage device.

As part of the data processing platform <NUM>, metadata <NUM> may be collected when changes are made to the data using a data manipulation language (DML), which changes may be made by way of any DML statement. Examples of manipulating data may include, but are not limited to, selecting, updating, changing, merging, and inserting data into tables. As part of the processing platform <NUM>, files may be created and the metadata <NUM> may be collected on a per file and a per column basis, after which the metadata <NUM> may be saved in a metadata store. This collection of metadata <NUM> may be performed during data ingestion or the collection of metadata <NUM> may be performed as a separate process after the data is ingested or loaded. In an implementation, the metadata <NUM> may include a number of distinct values; a number of null values; and a minimum value and a maximum value for each file. In an implementation, the metadata may further include string length information and ranges of characters in strings.

Resource manager <NUM> is further in communication with an execution platform <NUM>, which provides multiple computing resources that execute various data storage and data retrieval operations, as discussed in greater detail below. The execution platform <NUM> may include one or more compute clusters which may be dynamically allocated or suspended for specific warehouses, based on the query workload provided by the users <NUM>-<NUM> to a specific warehouse. The execution platform <NUM> is in communication with one or more data storage devices <NUM>, <NUM>, and <NUM> that are part of a storage platform <NUM>. Although three data storage devices <NUM>, <NUM>, and <NUM> are shown in <FIG>, the execution platform <NUM> is capable of communicating with any number of data storage devices. In some embodiments, data storage devices <NUM>, <NUM>, and <NUM> are cloud-based storage devices located in one or more geographic locations. For example, data storage devices <NUM>, <NUM>, and <NUM> may be part of a public cloud infrastructure or a private cloud infrastructure, or any other manner of distributed storage system. Data storage devices <NUM>, <NUM>, and <NUM> may include hard disk drives (HDDs), solid state drives (SSDs), storage clusters, or any other data storage technology. Additionally, the storage platform <NUM> may include a distributed file systems (such as Hadoop Distributed File Systems (HDFS)), object storage systems, and the like.

In some embodiments, the communication links between resource manager <NUM> and users <NUM>-<NUM>, metadata <NUM>, and execution platform <NUM> are implemented via one or more data communication networks and may be assigned various tasks such that user requests can be optimized. Similarly, the communication links between execution platform <NUM> and data storage devices <NUM>-<NUM> in storage platform <NUM> are implemented via one or more data communication networks. These data communication networks may utilize any communication protocol and any type of communication medium. In some embodiments, the data communication networks are a combination of two or more data communication networks (or sub-networks) coupled to one another. In alternate embodiments, these communication links are implemented using any type of communication medium and any communication protocol.

As shown in <FIG>, data storage devices <NUM>, <NUM>, and <NUM> are decoupled from the computing resources associated with execution platform <NUM>. This architecture supports dynamic changes to the data processing platform <NUM> based on the changing data storage/retrieval needs, computing needs, as well as the changing needs of the users and systems accessing data processing platform <NUM>. The support of dynamic changes allows data processing platform <NUM> to scale quickly in response to changing demands on the systems and components within data processing platform <NUM>. The decoupling of the computing resources from the data storage devices supports the storage of large amounts of data without requiring a corresponding large amount of computing resources. Similarly, this decoupling of resources supports a significant increase in the computing resources utilized at a particular time without requiring a corresponding increase in the available data storage resources.

The resource manager <NUM>, metadata <NUM>, execution platform <NUM>, and storage platform <NUM> are shown in <FIG> as individual components. However, each of resource manager <NUM>, metadata <NUM>, execution platform <NUM>, and storage platform <NUM> may be implemented as a distributed system (e.g., distributed across multiple systems/platforms at multiple geographic locations) or may be combined into one or more systems. Additionally, each of the resource manager <NUM>, storage for metadata <NUM>, the execution platform <NUM>, and the storage platform <NUM> can be scaled up or down (independently of one another) depending on changes to the requests received from users <NUM>-<NUM> and the changing needs of the data processing platform <NUM>. Thus, in the described embodiments, the data processing platform <NUM> is dynamic and supports regular changes to meet the current data processing needs.

The execution platform <NUM> includes a plurality of compute clusters <NUM>, <NUM>, <NUM> which may share a compute or processing load of the processing platform <NUM>. In one embodiment, customers can control the number of active (i.e. running) clusters by specifying a range (e.g., specifying values such as minClusterCount and maxClusterCount) when creating a warehouse or changing its configuration (both while the warehouse is running and while it is suspended). Customers may specify an exact number of active clusters by specifying, for example, making the minimum cluster count equal to the maximum cluster count so that the warehouse will have that exact number running whenever it is running. If a user specifies a maximum cluster count that is greater than a minimum cluster count, the resource manager <NUM> may automatically manage the number of currently active clusters based on the workload to satisfy the throughput criteria and to be cost-effective. So, whenever the warehouse is running, at least a minimum cluster count (minClusterCount) of clusters are active, and at most a maximum cluster count (maxClusterCount) of clusters are active. The resource manager <NUM> may decide how many clusters are required to handle the current workload given the specified performance criteria in terms of memory load and concurrency level.

<FIG> illustrates a block diagram depicting components of resource manager <NUM>, according to one embodiment. The resource manager <NUM> includes an access manager <NUM> and a key manager <NUM> coupled to a data storage device <NUM>. The access manager <NUM> handles authentication and authorization tasks for the systems described herein. The key manager <NUM> manages storage and authentication of keys used during authentication and authorization tasks. A request processing service <NUM> manages received data storage requests and data retrieval requests. A management console service <NUM> supports access to various systems and processes by administrators and other system managers.

The resource manager <NUM> also includes an SQL compiler <NUM>, an SQL optimizer <NUM> and an SQL executor <NUM>. SQL compiler <NUM> parses SQL queries and generates the execution code for the queries. SQL optimizer <NUM> determines the best method to execute queries based on the data that needs to be processed. SQL executor <NUM> executes the query code for queries received by resource manager <NUM>. A query scheduler and coordinator <NUM> sends received queries to the appropriate services or systems for compilation, optimization, and dispatch to an execution platform <NUM>. A virtual warehouse manager <NUM> manages the operation of multiple virtual warehouses, including multi-cluster warehouses, implemented in execution platform <NUM>.

Additionally, the resource manager <NUM> includes a configuration and metadata manager <NUM>, which manages the information related to the data stored in the remote data storage devices and in the local caches. A monitor and workload analyzer <NUM> oversees the processes performed by the resource manager <NUM> and manages the distribution of tasks (e.g., workload) across the virtual warehouses and execution nodes in the execution platform <NUM>. Configuration and metadata manager <NUM> and monitor and workload analyzer <NUM> are coupled to a data storage device <NUM>.

The resource manager <NUM> also includes a transaction management and access control module <NUM>, which manages the various tasks and other activities associated with the processing of data storage requests and data access requests. For example, the transaction management and access control module <NUM> provides consistent and synchronized access to data by multiple users or systems. Since multiple users/systems may access the same data simultaneously, changes to the data must be synchronized to ensure that each user/system is working with the current version of the data. Transaction management and access control module <NUM> provides control of various data processing activities at a single, centralized location in resource manager <NUM>.

With further reference to the virtual warehouse manager <NUM>, automatic cluster resume and automatic cluster suspend in a multi-cluster warehouse will be discussed. In one embodiment, the virtual warehouse manager <NUM> will perform automatic cluster resume. When a multi-cluster warehouse (e.g., within the execution platform <NUM>) is marked for automatic resume, the first cluster for the warehouse will be automatically resumed when a SQL statement is scheduled and all clusters in the warehouse are in a suspended state. But the decision to automatically resume the remaining clusters will be performed based on the workload. This assumes that activeClusterCount < maxClusterCount, that is, we have clusters that may be activated/resumed, but are currently suspended.

Workload considerations include at least two things. First, workload considerations may account for memory usage. When queries are scheduled and are queued because all clusters are at their maximum memory capacity, the virtual warehouse manager <NUM> will resume one or more clusters so that queueing can be avoided, or shortened. Queuing may still occur if new clusters need to be resumed since resuming a cluster may take a bit of time, for example in minutes. However, the virtual warehouse manager <NUM> may also make sure that there is a free pool of several free servers so that queries can be put on the free pool during the starting of the new cluster. Also, the virtual warehouse manager <NUM> may wait a specific period of time to see if queuing will resolve by itself before deciding to provision a new cluster.

Second, workload considerations may account for a degree of concurrence, or the processing/computing load on a cluster. If the degree of concurrency is high on all active clusters, then the virtual warehouse manager <NUM> may start another cluster even if there is enough memory to schedule the query. Here, the degree of concurrency may be computed for each active cluster based on the degree of parallelism (DOP). Specifically, the degree of concurrency may be the number of queries running at full DOP. For example, this may be calculated as the sum the DOP for all running queries divided by the product of a max DOP (MAX_DOP) and the number of running queries. This can be fractional or non-integer value since some lightweight queries are running with a smaller DOP than the max. In one embodiment, a warehouse parameter may be specified to control the degree of concurrency. For example, the degree of concurrency (concurrency_level_target_may be set to the value of <NUM> by default. This parameter may be exposed to a customer since its value really depends on how much money the customer wants to put on that problem and how much they are willing to allow query performance to degrade when a warehouse is shared (compared to the stand-alone query performance).

In one embodiment, the virtual warehouse manager <NUM> will perform automatic cluster suspend. In one embodiment, the full warehouse may shut down after a specific number of seconds (e.g., based on an auto_suspend parameter) of inactivity. Orthogonally to this, when the warehouse has more than one active cluster, one or more clusters may be suspended if the warehouse was running below its capacity for more than a specified amount of time, e.g. measured in minutes. For example, consider a warehouse with three active clusters. If for more than a specified time period the warehouse is underloaded, i.e. would have been able to execute all SQL statements issued in the specified time period engine at the current time without any queueing or without going over the maximum degree of concurrency, then one or more clusters would be suspended. Note that while a warehouse is underloaded, it will still leverage all clusters currently active. A cluster does not need to be inactive for a specific number of minutes to be shut down. A check for automatic cluster suspend may be performed on a periodic basis, such as for the last <NUM> minutes, last <NUM> minutes, last half hour, last hour, etc. In one embodiment, the check for automatic cluster may be performed at an interval different than the specified time period. For example, the check whether the last <NUM> minutes have been below load may be performed upon each hour change so that a customer can be charged on an hourly basis.

With further reference to the query scheduler and coordinator <NUM>, query scheduling may be performed based on workload, query affinity, and other factors. The query scheduler and coordinator <NUM> and may forward queries to a specific cluster based on workload. For example, the query scheduler and coordinator <NUM> may attempt to maintain an approximately equal workload on each cluster to spread out processing tasks and to improve query execution time and user experience. Query affinity may be used so that related queries, especially queries related to the same data, will be sent to the same cluster. For example, the query scheduler and coordinator <NUM> may send queries having a same session identifier to the same cluster. Forwarding queries based on query affinity may allow the query scheduler and coordinator <NUM> to ensure that the data against which a query is to be performed is already in the local cache of a specific cluster. This can significantly reduce response time, workload, and data lookup.

<FIG> is a schematic block diagram illustrating a multi-cluster warehouse <NUM> and the scheduling of queries <NUM> on the multi-cluster warehouse <NUM>. The warehouse <NUM> includes a plurality of clusters (Cluster <NUM>, Cluster <NUM>, Cluster N) that each include a plurality of server nodes. In one embodiment, each of the clusters includes the same number of servers although this may be different in different embodiments. In one embodiment, each server in a cluster belong to the same availability zone but different clusters may be placed in different availability zones. The concept of availability of the warehouse may be based on overall availability percentage of the warehouse. For example, the availability for a specific cluster within the warehouse <NUM> may be the percentage of servers which are available (e.g., in an operational state) relatively to the cluster size. However, when that percentage goes below the minimum (e.g., <NUM>%) required to run a query <NUM>% availability may be determined for that cluster and no queries may be assigned until the warehouse <NUM>, or some of the servers in the warehouse <NUM>, is repaired. As discussed herein, the number of clusters in the warehouse <NUM> may be adjusted dynamically based on workload, server failures in the clusters, or the like.

In one embodiment, the query scheduler and coordinator <NUM> weights each query (e.g., SQL statement or portion of a SQL statement) based on its projected resource consumption. For example, some queries may take significantly more memory to perform while other queries may take significantly more processing resources to perform. Similarly, some queries may have high or low consumption for both memory and processing. The resource manager <NUM> may determine what the predicted or projected consumption is and then may be able to determine where to place the query to most efficiently balance the workload among different clusters. For example, on high consumption query may use as many resources as multiple low consumption queries.

In one embodiment, the query scheduler and coordinator <NUM> may schedule queries on the one or more clusters of the warehouse <NUM> or may queue queries when workload is too high or availability is too low. For example, the query scheduler and coordinator <NUM> may first attempt to schedule a query <NUM> (e.g. a SQL statement) on an active (i.e. not suspended) cluster of the warehouse <NUM>. If there are multiple active clusters, the query scheduler and coordinator <NUM> will eliminate the set of clusters which are not available or where the query <NUM> would not run because memory would be oversubscribed. As mentioned previously, a cluster may be determined not available by default if less than <NUM>% of the servers of a node are not available (e.g., have failed). If there are multiple possible clusters remaining, the query scheduler and coordinator <NUM> may then pick the least loaded cluster. The least loaded cluster, in one embodiment, is defined as the sum of the DOPs of all jobs running on that cluster. The least loaded cluster may also be based on the sum of all memory requirements for that cluster. If there are multiple clusters with equal load, the query scheduler and coordinator <NUM> may use the session ID for the specific query <NUM> as a tie-breaker such that queries from the same session can execute on the same cluster. Queries <NUM> that have been assigned to a cluster in the warehouse <NUM> are shown as running queries <NUM>.

If there are not any clusters to schedule a specific query, then the query scheduler and coordinator <NUM> may queue the query in a global queue. Globally queued queries <NUM> are shown as queued queries <NUM>. Queued queries <NUM> may remain queued until one of the cluster of the warehouse <NUM> is freed up or becomes available. Note that one or more servers in an assigned cluster might be marked as suspected failed in which case some running queries <NUM> may also have to be queued waiting for the cluster to be repaired.

<FIG> is a block diagram depicting an embodiment of a plurality of warehouses which may be active or operating on a single execution platform <NUM>, according to one embodiment. Multiple virtual warehouses <NUM>, <NUM>, <NUM> are shown and each virtual warehouse includes a plurality of clusters <NUM>. Each cluster <NUM> includes multiple execution nodes <NUM> that each include a processor <NUM> and a cache <NUM> (e.g. memory). Although three virtual warehouses <NUM>-<NUM> are shown, the number of virtual warehouses may change dynamically. Similarly, the number of clusters <NUM> in each warehouse <NUM>-<NUM>, and the number of execution nodes <NUM> in each cluster may vary in different embodiments and may also vary relative to each other without limitation. Furthermore, the number of clusters <NUM> in a virtual warehouse and a number of execution nodes <NUM> in a cluster may be dynamic, such that new clusters <NUM> and execution nodes <NUM> are created or removed when demand changes.

Each virtual warehouse <NUM>-<NUM> is capable of accessing any of the data storage devices <NUM>-<NUM> shown in <FIG>. Thus, virtual warehouses <NUM>-<NUM> are not necessarily assigned to a specific data storage device <NUM>-<NUM> and, instead, can access data from any of the data storage devices <NUM>-<NUM>. Similarly, each of the clusters <NUM> and execution nodes <NUM> can access data from any of the data storage devices <NUM>-<NUM>. In some embodiments, a particular virtual warehouse or a particular execution node may be temporarily assigned to a specific data storage device, but the virtual warehouse or execution node may later access data from any other data storage device.

Although the illustrated execution nodes <NUM> each include one cache and one processor, alternate embodiments may include execution nodes containing any number of processors and any number of caches. Additionally, the caches may vary in size among the different execution nodes <NUM>. The caches <NUM> store, in the local execution node, data that was retrieved from one or more data storage devices in storage platform <NUM> (<FIG>). Thus, the caches reduce or eliminate the bottleneck problems occurring in platforms that consistently retrieve data from remote storage systems. Instead of repeatedly accessing data from the remote storage devices, the systems and methods described herein access data from the caches in the execution nodes which is significantly faster and avoids the bottleneck problem discussed above. In some embodiments, the caches are implemented using high-speed memory devices that provide fast access to the cached data. Each cache can store data from any of the storage devices in storage platform <NUM>.

Further, the cache resources and computing resources may vary between different execution nodes. For example, one execution node may contain significant computing resources and minimal cache resources, making the execution node useful for tasks that require significant computing resources. Another execution node may contain significant cache resources and minimal computing resources, making this execution node useful for tasks that require caching of large amounts of data. In some embodiments, the cache resources and computing resources associated with a particular execution node are determined when the execution node is created, based on the expected tasks to be performed by the execution node.

Additionally, the cache resources and computing resources associated with a particular execution node may change over time based on changing tasks performed by the execution node. For example, a particular execution node may be assigned more processing resources if the tasks performed by the execution node become more processor intensive. Similarly, an execution node may be assigned more cache resources if the tasks performed by the execution node require a larger cache capacity.

Although virtual warehouses <NUM>-<NUM> are associated with the same execution platform <NUM>, the virtual warehouses may be implemented using multiple computing systems at multiple geographic locations. For example, virtual warehouse <NUM> can be implemented by a computing system at a first geographic location, while virtual warehouses <NUM> and <NUM> are implemented by another computing system at a second geographic location. In some embodiments, these different computing systems are cloud-based computing systems maintained by one or more different entities.

Additionally, each virtual warehouse is shown in <FIG> as having multiple clusters <NUM>. The clusters <NUM> associated with each virtual warehouse may be implemented using multiple computing systems at multiple geographic locations or within different availability zones. For example, a particular instance of virtual warehouse <NUM> implements clusters <NUM> with execution nodes <NUM> on one computing platform at a particular geographic location, and implements other clusters <NUM> and execution nodes <NUM> at a different computing platform at another geographic location. The virtual warehouses <NUM>-<NUM> are also fault tolerant. For example, if one virtual warehouse or an execution node <NUM>, that virtual warehouse or execution node is quickly replaced at the same or different geographic location.

A particular execution platform <NUM> may include any number of virtual warehouses <NUM>-<NUM>. Additionally, the number of virtual warehouses in a particular execution platform is dynamic, such that new virtual warehouses are created when additional processing and/or caching resources are needed. Similarly, existing virtual warehouses may be deleted when the resources associated with the virtual warehouse are no longer necessary.

<FIG> illustrates a block diagram depicting another example operating environment <NUM> having multiple distributed virtual warehouses and execution platform groups. Environment <NUM> includes resource manager <NUM> that communicates with execution platform group <NUM><NUM> and execution platform group <NUM><NUM> through a data communication network <NUM>. Execution platform group <NUM><NUM> includes two clusters, specifically, cluster A for a first virtual warehouse <NUM> and cluster A for a second virtual warehouse <NUM>. Execution platform group <NUM><NUM> includes two additional clusters, specifically, cluster B for the first virtual warehouse <NUM> and cluster B for the second virtual warehouse <NUM>. The resource manager <NUM> also communicates with cluster C of the first virtual warehouse <NUM> (which is not part of either of the execution platform groups <NUM>, <NUM>) through data communication network <NUM>.

Execution platform groups <NUM> and <NUM> as well as cluster C for the first virtual warehouse <NUM> communicate with databases <NUM>, <NUM>, and <NUM> through a data communication network <NUM>. In some embodiments data communication networks <NUM> and <NUM> are the same network or a combination of one or more overlapping networks. Environment <NUM> allows resource manager <NUM> to coordinate user data storage and retrieval requests across multiple clusters <NUM>-<NUM> of multiple warehouses to store and retrieve data in databases <NUM>-<NUM>. Execution platform groups <NUM> and <NUM>, as well as cluster C for the first virtual warehouse512, can be located in the same or different geographic area, or can be located in the same or different availability zones. Additionally, execution platform groups <NUM> and <NUM> can be implemented by the same entity or by different entities.

The systems and methods described herein allow data to be stored and accessed as a service that is separate from computing (or processing) resources. Even if no computing resources have been requested from the execution platform, data is available to a virtual warehouse without requiring reloading of the data from a remote data source. The described systems and methods are useful with any type of data. In particular embodiments, data is stored in a structured, optimized format. The decoupling of the data storage/access service from the computing services also simplifies the sharing of data among different users and groups. As discussed herein, each virtual warehouse can access any data to which it has access permissions, even at the same time as other virtual warehouses are accessing the same data. This architecture supports running queries without any actual data stored in the local cache. The systems and methods described herein are capable of transparent dynamic data movement, which moves data from a remote storage device to a local cache, as needed, in a manner that is transparent to the user of the system. Further, this architecture supports data sharing without prior data movement since any virtual warehouse can access any data due to the decoupling of the data storage service from the computing service.

Furthermore, the environment <NUM> allows for the spreading of a single virtual warehouse across multiple geographic locations or availability zones. For example, clusters <NUM>, <NUM> and <NUM> all belong to the same virtual warehouse (the first virtual warehouse) but may be located in different geographic areas or availability zones. Because outages or failures may happen across a geographic area or availability zone, improved fault tolerance may be achieved. For example, availability zones are sometimes implemented by cloud service (compute or storage) providers so that problems in one availability zone have little or no chance of propagating to a different availability zone. Thus, clusters within the same warehouse but in different availability zones can significantly decrease the likelihood that a warehouse is left without any available execution or compute nodes.

In one embodiment, the multi-cluster warehouse embodiments disclosed herein may use a special data definition language (DDL). The following are some examples of commands or instructions which may be part of a multi-cluster warehouse DDL:.

Returning to the figures, <FIG> is a schematic flow chart diagram illustrating an example method <NUM> for a multi-cluster warehouse. The method <NUM> may be performed by a processing platform or a resource manager, such as the processing platform <NUM> of <FIG> or the resource manager of <FIG>, <FIG>, or <FIG>.

The method <NUM> begins and a system allocates <NUM> a plurality of compute clusters as part of a virtual warehouse for accessing and performing queries against one or more databases in one or more cloud storage resources. In one embodiment, the plurality of compute clusters is allocated by the system independently from the one or more cloud storage resources such that the number of compute clusters can be scaled up and down without increasing or decreasing the one or more cloud storage resources. The system provides <NUM> queries for the virtual warehouse to each of the plurality of compute clusters. For example, the plurality of queries may be provided to each of the plurality of compute clusters of the virtual warehouse. The system dynamically adds <NUM> compute clusters to and removes compute clusters from the virtual warehouse as needed based on a workload of the plurality of compute clusters. The method <NUM> may also include determining the workload for the plurality of compute clusters. The system may determine the workload by determining an availability of one or more of processor resources for each of the plurality of compute clusters and memory resources for each of the plurality of compute clusters.

Method <NUM> may be implemented by a database system or device to allow a single entity, such as a warehouse, to expand and shrink depending on a number of queries. Specifically, as changes in the concurrency (or compute and memory load) of a warehouse occur, a resource manager or other system may allow the warehouse to scale up and down.

<FIG> is a schematic flow chart diagram illustrating an example method <NUM> for dynamically adding compute clusters in a multi-cluster warehouse. The method <NUM> may be performed by a processing platform or a resource manager, such as the processing platform <NUM> of <FIG> or the resource manager of <FIG>, <FIG>, or <FIG>. The method <NUM> may be performed in conjunction with or separately from method <NUM> of <FIG>.

The method <NUM> begins and a system determines <NUM> whether a query can be processed while meeting a performance metric for the query. In one embodiment, the method <NUM> includes determining <NUM> whether the query can be processed for each query directed to the compute cluster such that the performance metric is met for each query. The performance metric may include a service level agreement (SLA) accepted by a customer. For example, the SLA may require that a query be scheduled within a specific amount of time (e.g., <NUM> seconds). This may restrict any query from being queued in a global queue longer than a maximum time (e.g., <NUM> seconds). The SLA may be agreed to in advance between a warehouse as a service provider and a customer. Different price tiers may be presented based on what the SLA is, or the SLA may dictate that the system uses more resources to ensure that users experience minimum delay in accessing and performing queries against a database.

The system triggers <NUM> startup of a new compute cluster in response to determining that the query in combination with a current workload does not allow one or more currently allocated compute clusters to meet the performance metric. In one embodiment, the system may only trigger <NUM> startup if the number of currently active clusters is less than a predetermined maximum number of compute clusters.

<FIG> is a schematic flow chart diagram illustrating an example method <NUM> for dynamically removing compute clusters in a multi-cluster warehouse. The method <NUM> may be performed by a processing platform or a resource manager, such as the processing platform <NUM> of <FIG> or the resource manager of <FIG>, <FIG>, or <FIG>. The method <NUM> may be performed in conjunction with or separately from one or more of methods <NUM> or <NUM> of <FIG>.

The method <NUM> begins and a system determines <NUM> whether a current workload is serviceable by one fewer than the plurality of compute clusters while meeting a performance metric. In one embodiment, determining <NUM> whether the current workload is serviceable by one fewer than the plurality of compute clusters may include determining whether a historical workload for a time period leading up to the current time was serviceable by one fewer than the plurality of clusters while meeting the performance metric. For example, if the best cluster were removed from the virtual warehouse, would the virtual warehouse have been able to process all the queries while meeting the performance metric?.

The system decommissions <NUM> (or inactivating) at least one compute cluster of the plurality of compute clusters in response to determining that the workload is serviceable by one fewer than the plurality of compute clusters. The system may only decommission <NUM> or remove a compute cluster if the current number of active clusters is less than a predetermined minimum number of compute clusters. In one embodiment, decommissioning <NUM> the at least one compute cluster may include decommissioning in response to determining that the historical workload for the time period was serviceable by one fewer than the plurality of compute clusters.

In one embodiment, decommissioning <NUM> the at least one compute cluster includes making a cluster quiescent to prevent providing or scheduling of additional queries to the at least one compute cluster. Decommissioning <NUM> may also include allowing the at least one compute cluster to complete currently assigned queries and releasing one or more resources corresponding to the at least one compute cluster upon completion of the already scheduled or active queries.

<FIG> is a schematic flow chart diagram illustrating an example method <NUM> for a multi-cluster warehouse. The method <NUM> may be performed by a processing platform or a resource manager, such as the processing platform <NUM> of <FIG> or the resource manager of <FIG>, <FIG>, or <FIG>. The method <NUM> may be performed in conjunction with or separately from one or more of methods <NUM>, <NUM>, or <NUM> of <FIG>, and <FIG>.

The method <NUM> begins and a system allocates <NUM> a plurality of compute clusters as part of a virtual warehouse for accessing and performing queries against one or more databases in one or more cloud storage resources. The system forwards <NUM> queries for the virtual warehouse to each of the plurality of compute clusters. The plurality of queries may be provided to each of the plurality of compute clusters of the virtual warehouse. In one embodiment, forwarding <NUM> queries for the virtual warehouse to each of the plurality of compute clusters includes routing <NUM> queries based on a session from which the query originated, such that queries from the same session are routed to a same compute cluster by default. Each cluster has the ability to persist fragments of the database it is operating on. That is, each cluster (or each compute node in the cluster) may maintain a cache of all the tables that it has currently accessed while processing queries on a cluster. Thus, the resource manager or scheduler drives the queries from the same query stream (e.g., having the same session identifier) to the same cluster so they can leverage the caching effect. In some cases, if a cluster that is handling a specific session has much less available resources than another cluster, queries with the same session identifier may end up on different clusters.

In one embodiment, the system may route <NUM> the queries based on a workload of each of the plurality of compute clusters. For example, if a cluster cannot accept new queries, the system may provide the query to a different cluster even if the different cluster has not processed queues corresponding to the same session. In one embodiment, the system may provide <NUM> queries to at least two compute clusters in different availability zones.

<FIG> is a block diagram depicting an example computing device <NUM>. In some embodiments, computing device <NUM> is used to implement one or more of the systems and components discussed herein. For example, computing device <NUM> may allow a user or administrator to access resource manager <NUM>. As another example, the components, systems, or platforms discussed herein may include one or more computing devices <NUM>. Further, computing device <NUM> may interact with any of the systems and components described herein. Accordingly, computing device <NUM> may be used to perform various procedures and tasks, such as those discussed herein. Computing device <NUM> can function as a server, a client or any other computing entity. Computing device <NUM> can be any of a wide variety of computing devices, such as a desktop computer, a notebook computer, a server computer, a handheld computer, a tablet, and the like.

Computing device <NUM> includes one or more processor(s) <NUM>, one or more memory device(s) <NUM>, one or more interface(s) <NUM>, one or more mass storage device(s) <NUM>, and one or more Input/Output (I/O) device(s) <NUM>, all of which are coupled to a bus <NUM>. Processor(s) <NUM> include one or more processors or controllers that execute instructions stored in memory device(s) <NUM> and/or mass storage device(s) <NUM>. Processor(s) <NUM> may also include various types of computer-readable media, such as cache memory.

Memory device(s) <NUM> include various computer-readable media, such as volatile memory (e.g., random access memory (RAM)) and/or nonvolatile memory (e.g., read-only memory (ROM)). Memory device(s) <NUM> may also include rewritable ROM, such as Flash memory.

Mass storage device(s) <NUM> include various computer readable media, such as magnetic tapes, magnetic disks, optical disks, solid state memory (e.g., Flash memory), and so forth. Various drives may also be included in mass storage device(s) <NUM> to enable reading from and/or writing to the various computer readable media. Mass storage device(s) <NUM> include removable media and/or non-removable media.

I/O device(s) <NUM> include various devices that allow data and/or other information to be input to or retrieved from computing device <NUM>. Example I/O device(s) <NUM> include cursor control devices, keyboards, keypads, microphones, monitors or other display devices, speakers, printers, network interface cards, modems, lenses, CCDs or other image capture devices, and the like.

Interface(s) <NUM> include various interfaces that allow computing device <NUM> to interact with other systems, devices, or computing environments. Example interface(s) <NUM> include any number of different network interfaces, such as interfaces to local area networks (LANs), wide area networks (WANs), wireless networks, and the Internet.

Bus <NUM> allows processor(s) <NUM>, memory device(s) <NUM>, interface(s) <NUM>, mass storage device(s) <NUM>, and I/O device(s) <NUM> to communicate with one another, as well as other devices or components coupled to bus <NUM>. Bus <NUM> represents one or more of several types of bus structures, such as a system bus, PCI bus, IEEE <NUM> bus, USB bus, and so forth.

The flow diagrams and block diagrams herein illustrate the architecture, functionality, and operation of possible implementations of systems, methods, and computer program products according to various embodiments of the present disclosure. In this regard, each block in the flow diagrams or block diagrams may represent a module, segment, or portion of code, which comprises one or more executable instructions for implementing the specified logical function(s). It will also be noted that each block of the block diagrams and/or flow diagrams, and combinations of blocks in the block diagrams and/or flow diagrams, may be implemented by special purpose hardware-based systems that perform the specified functions or acts, or combinations of special purpose hardware and computer instructions. These computer program instructions may also be stored in a computer-readable medium that can direct a computer or other programmable data processing apparatus to function in a particular manner, such that the instructions stored in the computer-readable medium produce an article of manufacture including instruction means which implement the function/act specified in the flow diagram and/or block diagram block or blocks.

The systems and methods described herein provide a flexible and scalable data warehouse using new data processing platforms, methods, systems, and algorithms. In some embodiments, the described systems and methods leverage a cloud infrastructure that supports cloud-based storage resources, computing resources, and the like. Example cloud-based storage resources offer significant storage capacity available on-demand at a low cost. Further, these cloud-based storage resources may be fault-tolerant and highly scalable, which can be costly to achieve in private data storage systems. Example cloud-based computing resources are available on-demand and may be priced based on actual usage levels of the resources. Typically, the cloud infrastructure is dynamically deployed, reconfigured, and decommissioned in a rapid manner.

In the described systems and methods, a data storage system utilizes an SQL (Structured Query Language)-based relational database. However, these systems and methods are applicable to any type of database using any data storage architecture and using any language to store and retrieve data within the database. The systems and methods described herein may also provide a multi-tenant system that supports isolation of computing resources and data between different customers/clients and between different users within the same customer/client.

Various techniques, or certain aspects or portions thereof, may take the form of program code (i.e., instructions) embodied in tangible media, such as floppy diskettes, CD-ROMs, hard drives, a non-transitory computer readable storage medium, or any other machine readable storage medium wherein, when the program code is loaded into and executed by a machine, such as a computer, the machine becomes an apparatus for practicing the various techniques. In the case of program code execution on programmable computers, the computing device may include a processor, a storage medium readable by the processor (including volatile and non-volatile memory and/or storage elements), at least one input device, and at least one output device. The volatile and non-volatile memory and/or storage elements may be a RAM, an EPROM, a flash drive, an optical drive, a magnetic hard drive, or another medium for storing electronic data. One or more programs that may implement or utilize the various techniques described herein may use an application programming interface (API), reusable controls, and the like. Such programs may be implemented in a high-level procedural or an object-oriented programming language to communicate with a computer system. However, the program(s) may be implemented in assembly or machine language, if desired. In any case, the language may be a compiled or interpreted language, and combined with hardware implementations.

It should be understood that many of the functional units described in this specification may be implemented as one or more components, which is a term used to more particularly emphasize their implementation independence. For example, a component may be implemented as a hardware circuit comprising custom very large scale integration (VLSI) circuits or gate arrays, off-the-shelf semiconductors such as logic chips, transistors, or other discrete components. A component may also be implemented in programmable hardware devices such as field programmable gate arrays, programmable array logic, programmable logic devices, or the like.

Components may also be implemented in software for execution by various types of processors. An identified component of executable code may, for instance, comprise one or more physical or logical blocks of computer instructions, which may, for instance, be organized as an object, a procedure, or a function. Nevertheless, the executables of an identified component need not be physically located together, but may comprise disparate instructions stored in different locations that, when joined logically together, comprise the component and achieve the stated purpose for the component.

Indeed, a component of executable code may be a single instruction, or many instructions, and may even be distributed over several different code segments, among different programs, and across several memory devices. Similarly, operational data may be identified and illustrated herein within components, and may be embodied in any suitable form and organized within any suitable type of data structure. The operational data may be collected as a single data set, or may be distributed over different locations including over different storage devices, and may exist, at least partially, merely as electronic signals on a system or network. The components may be passive or active, including agents operable to perform desired functions.

Claim 1:
A system comprising:
means for allocating a plurality of compute clusters (<NUM>-<NUM>, <NUM>) on an execution platform (<NUM>) as part of a virtual warehouse (<NUM>-<NUM>) for accessing and performing queries against one or more databases in one or more cloud storage resources located on a storage platform (<NUM>) separate from the execution platform, wherein the plurality of compute clusters is allocated separately from the one or more cloud storage resources (<NUM>-<NUM>) and each of the plurality of compute clusters comprises one or more execution nodes comprising a cache memory to cache data stored on the storage platform;
means for determining a current workload of the plurality of compute clusters based at least in part on a comparison of a runtime computed degree of concurrency on each of the plurality of compute clusters and a targeted degree of concurrency inputted by a customer; and
means for dynamically adding compute clusters to or removing compute clusters from the virtual warehouse based at least in part on whether one or more currently allocated compute clusters can process one or more of a query and the current workload while meeting a performance metric, the performance metric comprising a maximum query response time and wherein the means for adding or removing the compute clusters dynamically scales up and down a number of compute clusters without increasing or decreasing the one or more cloud storage resources.