Patent ID: 12229137

The figures depict various embodiments of the present invention for purposes of illustration only. One skilled in the art will readily recognize from the following discussion that alternative embodiments of the structures and methods illustrated herein may be employed without departing from the principles described herein.

DETAILED DESCRIPTION

The Figures (FIGS.) and the following description relate to preferred embodiments by way of illustration only. It should be noted that from the following discussion, alternative embodiments of the structures and methods disclosed herein will be readily recognized as viable alternatives that may be employed without departing from the principles of what is claimed.

Reference will now be made in detail to several embodiments, examples of which are illustrated in the accompanying figures. It is noted that wherever practicable similar or like reference numbers may be used in the figures and may indicate similar or like functionality. The figures depict embodiments of the disclosed system (computer-readable medium or method) for purposes of illustration only. One skilled in the art will readily recognize from the following description that alternative embodiments of the structures and methods illustrated herein may be employed without departing from the principles described herein.

Overview

The disclosed configurations provide a method (and/or a computer-readable medium or computer system) for efficient startup of executors of a distributed computing engine used for processing queries, for example, database queries. The system starts an executor node and processes a set of queries using the executor node. The processing of the queries causes the necessary instructions, for example, classes including code for processing various types of queries to be loaded in memory, variables to be initialized, data structures to be created, and code compiled to generate native instructions if needed. The system performs a checkpoint of the executor node to create a checkpoint image. The system sends the checkpoint image to various target executor nodes. The image is restored in the target executor nodes. Since the checkpoint image was created using a pre-warmed executor node that loaded various types of data that is created or loaded at startup time, the target executor nodes preload the necessary information that is created during startup. As a result, the startup of the target executor nodes is significantly faster than the startup of a typical executor node using conventional techniques. For example, an improvement of an order of magnitude was experimentally observed wherein an executor node using the disclosed techniques performed the startup in few seconds whereas startup using conventional techniques took a couple of minutes. A checkpoint image may also be referred to herein as an image, or an executor image.

The data processing service102is a service for managing and coordinating data processing services (e.g., database services) to users of client devices116. The data processing service102may manage one or more applications that users of client devices116can use to communicate with the data processing service102. Through an application of the data processing service102, the data processing service102may receive requests (e.g., database queries) from users of client devices116to perform one or more data processing functionalities on data stored, for example, in the data storage system110. The requests may include query requests, analytics requests, or machine learning and artificial intelligence requests, and the like, on data stored by the data storage system110. The data processing service102may provide responses to the requests to the users of the client devices116after they have been processed.

In one embodiment, as shown in the system environment100ofFIG.1, the data processing service102includes a control layer106and a data layer108. The components of the data processing service102may be configured by one or more servers and/or a cloud infrastructure platform. In one embodiment, the control layer106receives data processing requests and coordinates with the data layer108to process the requests from client devices116. The control layer106may schedule one or more jobs for a request or receive requests to execute one or more jobs from the user directly through a respective client device116. The control layer106may distribute the jobs to components of the data layer108where the jobs are executed.

The control layer106is additionally capable of configuring the clusters in the data layer108that are used for executing the jobs. For example, a user of a client device116may submit a request to the control layer106to perform one or more queries and may specify that four clusters on the data layer108be activated to process the request with certain memory requirements. Responsive to receiving this information, the control layer106may send instructions to the data layer108to activate the requested number of clusters and configure the clusters according to the requested memory requirements.

The data layer108includes multiple instances of clusters of computing resources that execute one or more jobs received from the control layer106. Accordingly, the data layer108may include a cluster computing system for executing the jobs. An example of a cluster computing system is described in relation toFIG.4. In one instance, the clusters of computing resources are virtual machines or virtual data centers configured on a cloud infrastructure platform. In one instance, the control layer106is configured as a multi-tenant system and the data layers108of different tenants are isolated from each other. In one instance, a serverless implementation of the data layer108may be configured as a multi-tenant system with strong virtual machine (VM) level tenant isolation between the different tenants of the data processing service102. Each customer represents a tenant of a multi-tenant system and shares software applications and also resources such as databases of the multi-tenant system. Each tenant's data is isolated and remains invisible to other tenants. For example, a respective data layer instance can be implemented for a respective tenant. However, it is appreciated that in other embodiments, single tenant architectures may be used.

The data layer108thus may be accessed by, for example, a developer through an application of the control layer106to execute code developed by the developer. In one embodiment, a cluster in a data layer108may include multiple worker nodes that execute multiple jobs in parallel. Responsive to receiving a request, the data layer108divides the cluster computing job into a set of worker jobs, provides each of the worker jobs to a worker node, receives worker job results, stores job results, and the like. The data layer108may include resources not available to a developer on a local development system, such as powerful computing resources to process very large data sets. In this manner, when the data processing request can be divided into jobs that can be executed in parallel, the data processing request can be processed and handled more efficiently with shorter response and processing time.

The data storage system110includes a device (e.g., a disc drive, a hard drive, a semiconductor memory) used for storing database data (e.g., a stored data set, portion of a stored data set, data for executing a query). In one embodiment, the data storage system110includes a distributed storage system for storing data and may include a commercially provided distributed storage system service. Thus, the data storage system110may be managed by a separate entity than an entity that manages the data processing service102or the data storage system110may be managed by the same entity that manages the data processing service102.

The client devices116are computing devices that display information to users and communicates user actions to the systems of the system environment100. While two client devices116A,116B are illustrated inFIG.1, in practice many client devices116may communicate with the systems of the system environment100. In one embodiment, client devices116of the system environment100may include some or all of the components (systems (or subsystems)) of a computer system1000as described withFIG.10.

In one embodiment, a client device116executes an application allowing a user of the client device116to interact with the various systems of the system environment100ofFIG.1. For example, a client device116can execute a browser application to enable interaction between the client device116and the data processing service102via the network120. In another embodiment, the client device116interacts with the various systems of the system environment100through an application programming interface (API) running on a native operating system of the client device116, such as IOS® or ANDROID™

Data Storage System

FIG.2illustrates a block diagram of an architecture of the data storage system110, in accordance with an embodiment. As illustrated inFIG.2, the data storage system110includes a data store270and a metadata store275. The data storage system110may have multiple instances of the data store270and the metadata store275, each dedicated to storing data for a tenant of the data processing service102. In one embodiment, the data store270stores data in a format of a data table. A data table may include a set of records, where each record may include values for one or more features or keys. A feature may represent a measurable piece of data that can be used for analysis, such as, login account, timestep, etc. A feature may refer to any characteristic of the data that can be associated with one or more values of various data types, such as discrete or continuous values, string values, and the like. In one embodiment, the records may span across multiple rows of the data table and the features may span across multiple columns of the data table. In other embodiments, the records may span across multiple columns and the features may span across multiple rows. For example, a data table associated with a company may include a plurality of records each corresponding to a login instance of a respective user to a website, where each record includes values for a set of features including user login account, timestamp of attempted login, whether the login was successful, and the like.

In one embodiment, a data table may be stored in the data store270in conjunction with metadata stored in the metadata store275. As described above, the metadata store275may include one or more metadata files for a data table. In some embodiments, the metadata store275stores metadata in the form of one or more transaction logs. A transaction log for a data table is a log recording a sequence of transactions that were performed on the data table. A transaction may perform one or more changes to a data table that may include removal, modification, or addition of records and features to the data table, and the like. For example, a transaction may be initiated responsive to a request from a user of the client device116. As another example, a transaction may be initiated according to policies of the data processing service102. Thus, a transaction may write one or more changes to data tables stored in the data storage system110. The metadata in the transaction log may record removal, update, or addition of data files to a data table.

Control Layer

FIG.3is a block diagram of an architecture of a control layer106, in accordance with an embodiment. In one embodiment, the data processing service102includes an interface module325, a transaction module330, and a query processing module335. The modules325,330, and335may be structured for execution by a computer system, e.g.,1000having some or all of the components as described inFIG.10, such that the computer system1000operates in a specified manner as per the described functionality.

The interface module325provides an interface and/or a workspace environment where users of client devices116(e.g., users associated with tenants) can access resources of the data processing service102. For example, the user may retrieve information from data tables associated with a tenant, submit data processing requests such as query requests on the data tables, through the interface provided by the interface module325. The interface provided by the interface module325may include notebooks, libraries, experiments, queries submitted by the user. In one embodiment, a user may access the workspace via a user interface (UI), a command line interface (CLI), or through an application programming interface (API) provided by the interface module325.

The transaction module330receives requests to perform one or more transaction operations from users of client devices116. As described in conjunction inFIG.2, a request to perform a transaction operation may represent one or more requested changes to a data table. For example, the transaction may be to insert new records into an existing data table, replace existing records in the data table, delete records in the data table. As another example, the transaction may be to rearrange or reorganize the records or the data files of a data table to, for example, improve the speed of operations, such as queries, on the data table. For example, when a particular version of a data table has a significant number of data files composing the data table, some operations may be relatively inefficient. Thus, a transaction operation may be a compaction operation that combines the records included in one or more data files into a single data file.

The query processing module335receives and processes queries that access data stored by the data storage system110. The query processing module335may reside in the control layer106. The queries processed by the query processing module335are referred to herein as database queries. The database queries are specified using a declarative database query language such as the SQL. The query processing module335compiles a database query specified using the declarative database query language to generate executable code that is executed. The query processing module335may encounter runtime errors during execution of a database query and returns information describing the runtime error including an origin of the runtime error representing a position of the runtime error in the database query. In one embodiment, the query processing module335provides one or more queries to appropriate clusters of the data layer108and receives responses to the queries from clusters in which the queries are executed.

FIG.4is a block diagram of an architecture of a cluster computing system402of the data layer108, in accordance with an embodiment. In some embodiments, the cluster computing system402of the data layer108includes driver node450and worker pool including multiple executor nodes. The nodes may be structured for execution by a computer system, e.g.,1000having some or all of the components as described inFIG.10, such that the computer system1000operates in a specified manner as per the described functionality.

The driver node450receives one or more jobs for execution, divides a job into job stages, and provides job stages to executor nodes, receives job stage results from the executor nodes of the worker pool, and assembles job stage results into complete job results, and the like. In one embodiment, the driver node receives a request to execute one or more queries from the query processing module335. The driver node450may compile a database query and generate an execution plan. The driver node450distributes the query information including the generated code to the executor nodes. The executor nodes execute the query based on the received information.

The worker pool can include any appropriate number of executor nodes (e.g.,4executor nodes,12executor nodes,256executor nodes). Each executor node in the worker pool includes one or more execution engines (not shown) for executing one or more tasks of a job stage. In one embodiment, an execution engine performs single-threaded task execution in which a task is processed using a single thread of the CPU. The executor node distributes one or more tasks for a job stage to the one or more execution engines and provides the results of the execution to the driver node410. According to an embodiment, an executor node executes the generated code for the database query for a particular subset of data that is processed by the database query. The executor nodes execute the query based on the received information from the driver node450.

FIG.5is a block diagram of an architecture of a driver node450, in accordance with an embodiment. In one instance, the driver node450includes a query parser510, a query rewrite module520, a logical plan generation module530, a physical plan generation module540, a code generator module550, a checkpoint/restore module560, and a checkpoint image repository570. The modules and nodes may be structured for execution by a computer system, e.g.,1000having some or all of the components as described inFIG.10, such that the computer system1000operates in a specified manner as per the described functionality.

The query parser510receives a database query for processing and parses the database query. The database query is specified using a declarative database query language such as SQL. The query parser510parses the database query to identify various tokens of the database query and build a data structure representation of the database query. The data structure representation identifies various components of the database query, for example, any SELECT expressions that are returned by the database query, tables that are input to the query, a conditional clause of the database query, a group by clause, and so on.

The query rewrite module520performs transformations of the database query, for example, to improve the execution of the query. The improvement may be in terms of execution time, memory utilization, or other resource utilization. A database query may process one or more tables that store a significant number of records that are processed by the database query. Since the declarative database query language does not specify the procedure for determining the result of the database query, there are various possible procedures for executing the database query.

The logical plan generation module530generates a logical plan for the database query. The logical plan includes representation of the various steps that need to be executed for processing the database query. According to an embodiment, the logical plan generation module530generates an unresolved logical plan based on the transformed query graph representation. Various relation names (or table names) and column names may not be resolved in an unresolved logical plan. The logical plan generation module530generates a resolved logical plan from the unresolved logical plan by resolving the relation names and column names in the unresolved logical plan. The logical plan generation module530further optimizes the resolved logical plan to obtain an optimized logical plan.

The physical plan generation module540generates a physical plan from the logical plan generated by the logical plan generation module530. The physical plan specifies details of how the logical plan is executed by the data processing service102. The physical plan generation module540may generate different physical plans for the same logical plan and evaluate each physical plan using a cost model to select the optimal physical plan for execution. The physical plan further specifies details of various operations of the logical plan. As an example, if the logical plan includes a join operator, the physical plan may specify the type of join that should be performed for implementing the join operator.

The code generator module550generates code representing executable instructions for implementing the physical plan for executing a database query. The generated code includes a set of instructions for each operator specified in the execution plan. The generated code is specified using a programming language that may be compiled and executed.

The checkpoint/restore module560sends instructions to executor nodes for performing a checkpoint or a restore operation. The checkpoint operation results in creation of a checkpoint image. The checkpoint images created by the checkpoint operations are stored in the checkpoint image repository570. The checkpoint image repository570may be stored locally in the driver node450or may be implemented in a cloud platform. According to an embodiment, the checkpoint repository is stored on the executor node. For example, a checkpoint may be created on an executor node on-demand, and then repeatedly used to handle subsequent queries such that each query is processed by a freshly-restored container for security reasons.

The checkpoint/restore module560accesses the appropriate image for a given configuration of executor nodes from the checkpoint image repository570for performing a restore operation.

According to an embodiment, the executor node performs checkpoints of user level processes. For example, the executor node may run a set of user level processes within a container and the checkpoint/restore module560performs a checkpoint of all the user level processes running in the container. A container represents a unit of software that packages code for an application and all its dependencies so that the application runs efficiently in a computing environment.

A container is distinct from a virtual machine (VM) that emulates a physical hardware such that a single physical server can run multiple VMs, each acting as a distinct server. A VM includes a copy of an operating system, the application, necessary binaries and libraries, and so on. A container takes up less space than a VM. An image stores all the data needed to run the application including the code, libraries, settings, and so on. State of the art checkpoint images typically are tens of MBs in size as opposed to VM images that can be several GBs in size. As a result, container level checkpoint/restore operation is significantly more efficient compares to a VM level checkpoint/restore operation.

Furthermore, checkpointing user level processes provides higher level of security comparted to VM level checkpointing since kernel level data is not stored in the image and is therefore not restored in the executor nodes. This allows differences in execution of the container when restored in different executor nodes, for example, due to differences in random numbers generated by kernel level instructions. For example, random numbers may be used to generate secure keys used by the containers and if the same random numbers are generated by different instances of the container restored across multiple executor nodes, the same secure key may be generated across different instances of the container, thereby compromising security. Furthermore, an executor node may use address space randomization (ASLR) techniques for computer security that randomly arranges the address space positions of specific data areas of processes. VM level checkpoint/restore is likely to result in the same set of random numbers being generated across multiple executor nodes, thereby compromising the security offered by the ASLR techniques.

The code of executor node avoids host specific attributes such as hostname, IP addresses, and so on are avoided in the image, region specific information is avoided, for example, certificates used for authentication, time zone information, language specific information, and so on are avoided. The code of an executor node is structured so that the above information is stored in variables. After a restore operation is performed on an executor node, the above information is bound to their respective variables. If the executor code needs to access host information, the code is structured to access localhost rather than a specific host name. This results in generating a checkpoint image that is usable across executor nodes. For example, when the checkpoint image is restored on a new executor node, the hostname and IP addresses are available and can be accessed using appropriate variables.

Startup Based on Checkpoint/Restore of Executor Nodes

FIGS.6-8show flowcharts of processes used for performing efficient startup of an executor node, in accordance with an embodiment. The process shown inFIGS.6-8may be performed by one or more components (e.g., the checkpoint/restore module560of the driver node450) of a data processing system/service (e.g., the data processing service102). Other entities may perform some or all of the steps inFIG.6. The data processing service102as well as the other entities may include some or of the component of the machine (e.g., computer system) described in conjunction withFIG.10. Embodiments may include different and/or additional steps or perform the steps in different orders.

FIG.6shows a flowchart of a process for performing checkpoint of an executor node to generate an image, in accordance with an embodiment. The executor starts610processes for running queries, for example, the runtime environment for running instructions for processing queries. The queries may represent user requests that result in execution of database queries. The driver node450processes620a set of queries on the executor node. These queries include sample queries of various types, thereby exercising different parts of the code and invoke various possible libraries. The processing of queries causes the executor node to perform various operations as described in the flowchart shown inFIG.7and described in connection withFIG.7.

The checkpoint/restore module560performs630a checkpoint of the executor node. The checkpoint/restore module560may perform the checkpoint by sending instructions to the executor node to checkpoint the processes executing the queries, for example, the runtime engine executing the instructions such as the Java™ runtime. The checkpoint/restore module560stores640metadata describing the image generated. For example, the checkpoint/restore module560may generate multiple checkpoints, one for each type of hardware used by an executor node. The checkpoint/restore module560may store metadata describing the type of executor node for which the image was generated.

FIG.7shows a flowchart of a startup process of an executor node, in accordance with an embodiment. The steps shown inFIG.7may correspond to the step610of the flowchart shown inFIG.6. The steps shown inFIG.7may be performed in an order different from that indicated herein, for example, a step may be performed concurrently with other steps.

The executor node loads710classes and libraries comprising instructions for processing queries. Some classes and libraries may be loaded as the queries are executed by the executor node. For example, instructions for processing a specific query may invoke a function defined in a particular library, thereby causing that library to be loaded. A typical executor node may load thousands of libraries for processing queries. The executor node initializes720variables and data structures needed for processing queries. For example, some variables may represent state needed for processing the queries. Running the queries triggers the required initializations and creation of data structures.

The executor node further identifies hotspots in the code processing the queries and performs just-in-time compilation of the identified code to generate native executable instructions. For example, the code used for processing queries may be implemented using a language such as Java™ that generates byte code that is interpreted using a runtime environment. The execution of byte code may be slow since it is interpreted. As a result, the runtime environment of the executor node identifies frequently executed code as hotspots and further compile the hotspot code into native executable instructions. Execution of code compiled into native executable instructions is more efficient compared to the same code compiled into byte code and interpreted using a runtime engine. A typical executor node may generate several megabytes of machine code by performing just-in-time compilation of hotspots.

The image generated by the checkpoint operation stores the information generated during the process illustrated inFIG.7including the classes and libraries loaded, native code obtained by just-in-time compilation, variables, register values, files that were opened, any offsets determined in the files, state of each thread, state of the heap (represented as pages of the heap), state of the stack, and so on.

The steps illustrated inFIG.7cause the executor node to perform the processing that needs to be performed by each executor node. Accordingly, without using the technique disclosed herein, significant overhead (e.g., approximately two minutes of computational overhead) is incurred by each executor node during startup. Generating an image from the executor node that has already performed these causes the generated information to be stored in the image. This image can be restored in any target executor node and provides all the information and loading the information rather than having to perform the computation described in the process illustrated inFIG.7. Although the techniques are described in connection with executor nodes implemented using Java™ code, the techniques are applicable to any other implementation of executor nodes.

The executor node may perform optimizations to reduce the amount of memory that is stored in an image as a result of the checkpoint operation. For example, the executor node may remove one or more processes that occupy large memory after the queries are processed to reduce the heap memory. Furthermore, the executor node performs a garbage collection after processing the queries to remove unused pages from the memory. Reducing the size of the image improves the efficiency of storing the image as well as transmitting the image over network as well as improves the time taken to perform the checkpoint and restore operations.

FIG.8shows a flowchart of a process for performing restore of one or more executor nodes for performing efficient startup of the executor nodes, in accordance with an embodiment. The steps810,820, and830may be executed multiple times, for example, once for each target executor node being used for executing queries. The checkpoint/restore module560determines810the configuration (or the environment) of the executor node, for example, the type of hardware running the executor node. For example, different configurations of executor nodes may use different types of CPUs that support different instruction sets. Different executor nodes may use different amount of memory, for example, some executor nodes may allocate larger heap compared to other executor nodes. Each such executor node is treated as a different configuration for which a different image is generated. A different image is generated for each such configuration of executor nodes. The checkpoint/restore module560identifies820an image matching the configuration of the executor node. The checkpoint/restore module560restores830the image on the target executor node. The checkpoint/restore module560makes840the target executor node available for processing queries.

FIG.9shows a flowchart of a process of continuous integration/continuous delivery for images for new executor configurations, in accordance with an embodiment. The checkpoint/restore module560detects910a new configuration of executor nodes for which an image has not been previously created and stored. New configurations of executor nodes may be detected by querying information describing various executor nodes. The checkpoint/restore module560checks the various executor configuration attributes such as versions of various systems relevant to query processing that are installed on an executor node, environment attributes, hardware configuration, and so on. A signature based on the executor configuration may be generated and compared against signatures of executor nodes corresponding to existing images to determine whether the detected executor configuration is new (i.e., an executor configuration for which an image hasn't been generated so far and stored in the checkpoint image repository570).

The checkpoint/restore module560performs the processing shown inFIG.6to generate920a new image for the new configuration. The checkpoint/restore module560provides the details of the new configuration of the executor so that an example executor node is started and warmed up by processing images so that a new image for the new executor configuration can be created by checkpointing.

The new image is validated930. For example, certain tests may be executed using an executor started using the new image. The testing and validation ensure that the new image can be used to successfully startup an executor node having appropriate configuration and the new executor can be added to a cluster and used for processing queries. If the validation and testing is successful, the new image is promoted940for use in a production environment.

According to an embodiment, the checkpoint image repository maps each checkpoint image to a signature based on the various attributes that characterize the configuration of the executor node including hardware attributes, operating system attributes, and so on. When an executor node needs to be started, the checkpoint/restore module560determines the configuration of the executor node; determines a signature based on the configuration of the executor node; identifies a checkpoint image mapped to that signature; and uses the checkpoint image to restore the executor node.

According to an embodiment, the executor node tracks all random numbers used by the code. The random numbers are regenerated after the image is restored on the executor node, for example, the random number value may be replaced with new random values generated on executor node on which the restore was performed. This way different executor nodes restored using the same image are able to use different set of random numbers.

The checkpoint/restore module560may implement a restore operation for executor nodes that loads all pages in memory when the restore operation in performed. In some embodiments, the system implements a restore operation that lazily loads pages as they are needed while processing queries. According to some embodiments, the checkpoint/restore module560detects whether there are new configurations of executor nodes and generates corresponding images.

According to an embodiment, the checkpoint/restore module560generates multiple checkpoint images for each distinct configuration of executor node. The different checkpoint images have different address space layouts by performing address space layout randomization. For example, the checkpoint/restore module560may generate64distinct images each image having a different address space layout. The checkpoint/restore module560may use different restore strategies for using the different checkpoint images for the same configuration of the executor node. For example, the checkpoint/restore module560may use a particular image to restore a particular executor and after a predetermined interval of time, shutdown the executor node and restore a different image. As another strategy, the checkpoint/restore module560uses a new image for the executor, every time the executor needs to be restarted, for example, as a result of container crash. The checkpoint/restore module560may switch to a different image for an executor node if a suspicious activity is detected using that executor node, for example, an attempt to access sensitive information on that executor node. According to an embodiment, if a suspicious activity is detected, the executor node is restarted using conventional startup process that does not restore an image but runs the executable instructions of each user level process to start the process.

According to an embodiment, an executor node performs runtime re-randomization of the address space after the checkpoint image is restored. During the checkpoint stage, the executor code is run using a CPU emulator. During the warm-up process when the queries are executed using the executor node, the emulated CPU instructions are instrumented to track how pointers are propagated in the process' address space. When restoring the checkpoint, the executor node reshuffles the memory segments (e.g., library images, stack and heap) and fixes up all the tracked pointers so that they point to the new locations within the address space.

Turning now toFIG.10, illustrated is an example machine to read and execute computer readable instructions, in accordance with an embodiment. Specifically,FIG.10shows a diagrammatic representation of the data processing service102(and/or data processing system) in the example form of a computer system1000. The computer system1000can be used to execute instructions1024(e.g., program code or software) for causing the machine to perform any one or more of the methodologies (or processes) described herein. In alternative embodiments, the machine operates as a standalone device or a connected (e.g., networked) device that connects to other machines. In a networked deployment, the machine may operate in the capacity of a server machine or a client machine in a server-client network environment, or as a peer machine in a peer-to-peer (or distributed) network environment.

The machine may be a server computer, a client computer, a personal computer (PC), a tablet PC, a set-top box (STB), a smartphone, an internet of things (IoT) appliance, a network router, switch or bridge, or any machine capable of executing instructions824(sequential or otherwise) that specify actions to be taken by that machine. Further, while only a single machine is illustrated, the term “machine” shall also be taken to include any collection of machines that individually or jointly execute instructions1024to perform any one or more of the methodologies discussed herein.

The example computer system1000includes one or more processing units (generally processor1002). The processor1002is, for example, a central processing unit (CPU), a graphics processing unit (GPU), a digital signal processor (DSP), a controller, a state machine, one or more application specific integrated circuits (ASICs), one or more radio-frequency integrated circuits (RFICs), or any combination of these. The processor executes an operating system for the computing system800. The computer system1000also includes a main memory1004. The computer system may include a storage unit1016. The processor1002, main memory1004, and the storage unit1016communicate via a bus1008.

In addition, the computer system1000can include a static memory1006, a graphics display1010(e.g., to drive a plasma display panel (PDP), a liquid crystal display (LCD), or a projector). The computer system1000may also include alphanumeric input device1012(e.g., a keyboard), a cursor control device1014(e.g., a mouse, a trackball, a joystick, a motion sensor, or other pointing instrument), a signal generation device1018(e.g., a speaker), and a network interface device1020, which also are configured to communicate via the bus1008.

The storage unit1016includes a machine-readable medium1022on which is stored instructions1024(e.g., software) embodying any one or more of the methodologies or functions described herein. For example, the instructions1024may include instructions for implementing the functionalities of the transaction module330. The instructions1024may also reside, completely or at least partially, within the main memory1004or within the processor1002(e.g., within a processor's cache memory) during execution thereof by the computer system1000, the main memory1004and the processor1002also constituting machine-readable media. The instructions1024may be transmitted or received over a network1026, such as the network120, via the network interface device1020.

While machine-readable medium1022is shown in an example embodiment to be a single medium, the term “machine-readable medium” should be taken to include a single medium or multiple media (e.g., a centralized or distributed database, or associated caches and servers) able to store the instructions1024. The term “machine-readable medium” shall also be taken to include any medium that is capable of storing instructions1024for execution by the machine and that cause the machine to perform any one or more of the methodologies disclosed herein. The term “machine-readable medium” includes, but not be limited to, data repositories in the form of solid-state memories, optical media, and magnetic media.

Additional Configuration Considerations

The disclosed configurations beneficially provide a method (and/or a system) for resolving concurrent transactions using deletion vectors. Without deletion vectors, concurrent transactions may fail to commit due to the conflicts introduced by intervening transactions, and the current transaction typically will have to start over from the most current version of the data table, resulting in wasted computing resources and time. By resolving concurrent conflicts in conjunction with deletion vectors associated with data files, the method presented herein may determine the serializability of the concurrent transactions and prevent many transaction operations from failing and having to be started over again, saving resources and cost for both the data processing service and the users of the data processing service.

Throughout this specification, plural instances may implement components, operations, or structures described as a single instance. Although individual operations of one or more methods are illustrated and described as separate operations, one or more of the individual operations may be performed concurrently, and nothing requires that the operations be performed in the order illustrated. Structures and functionality presented as separate components in example configurations may be implemented as a combined structure or component. Similarly, structures and functionality presented as a single component may be implemented as separate components. These and other variations, modifications, additions, and improvements fall within the scope of the subject matter herein.

Certain embodiments are described herein as including logic or a number of components, modules, or mechanisms. Modules may constitute either software modules (e.g., code embodied on a machine-readable medium or in a transmission signal) or hardware modules. A hardware module is tangible unit capable of performing certain operations and may be configured or arranged in a certain manner. In example embodiments, one or more computer systems (e.g., a standalone, client or server computer system) or one or more hardware modules of a computer system (e.g., a processor or a group of processors) may be configured by software (e.g., an application or application portion) as a hardware module that operates to perform certain operations as described herein.

In various embodiments, a hardware module may be implemented mechanically or electronically. For example, a hardware module may comprise dedicated circuitry or logic that is permanently configured (e.g., as a special-purpose processor, such as a field programmable gate array (FPGA) or an application-specific integrated circuit (ASIC)) to perform certain operations. A hardware module may also comprise programmable logic or circuitry (e.g., as encompassed within a general-purpose processor or other programmable processor) that is temporarily configured by software to perform certain operations. It will be appreciated that the decision to implement a hardware module mechanically, in dedicated and permanently configured circuitry, or in temporarily configured circuitry (e.g., configured by software) may be driven by cost and time considerations.

Accordingly, the term “hardware module” should be understood to encompass a tangible entity, be that an entity that is physically constructed, permanently configured (e.g., hardwired), or temporarily configured (e.g., programmed) to operate in a certain manner or to perform certain operations described herein. As used herein, “hardware-implemented module” refers to a hardware module. Considering embodiments in which hardware modules are temporarily configured (e.g., programmed), each of the hardware modules need not be configured or instantiated at any one instance in time. For example, where the hardware modules comprise a general-purpose processor configured using software, the general-purpose processor may be configured as respective different hardware modules at different times. Software may accordingly configure a processor, for example, to constitute a particular hardware module at one instance of time and to constitute a different hardware module at a different instance of time.

Hardware modules can provide information to, and receive information from, other hardware modules. Accordingly, the described hardware modules may be regarded as being communicatively coupled. Where multiple of such hardware modules exist contemporaneously, communications may be achieved through signal transmission (e.g., over appropriate circuits and buses) that connect the hardware modules. In embodiments in which multiple hardware modules are configured or instantiated at different times, communications between such hardware modules may be achieved, for example, through the storage and retrieval of information in memory structures to which the multiple hardware modules have access. For example, one hardware module may perform an operation and store the output of that operation in a memory device to which it is communicatively coupled. A further hardware module may then, at a later time, access the memory device to retrieve and process the stored output. Hardware modules may also initiate communications with input or output devices, and can operate on a resource (e.g., a collection of information).

The various operations of example methods described herein may be performed, at least partially, by one or more processors that are temporarily configured (e.g., by software) or permanently configured to perform the relevant operations. Whether temporarily or permanently configured, such processors may constitute processor-implemented modules that operate to perform one or more operations or functions. The modules referred to herein may, in some example embodiments, comprise processor-implemented modules.

Similarly, the methods described herein may be at least partially processor-implemented. For example, at least some of the operations of a method may be performed by one or processors or processor-implemented hardware modules. The performance of certain of the operations may be distributed among the one or more processors, not only residing within a single machine, but deployed across a number of machines. In some example embodiments, the processor or processors may be located in a single location (e.g., within a home environment, an office environment or as a server farm), while in other embodiments the processors may be distributed across a number of locations.

The one or more processors may also operate to support performance of the relevant operations in a “cloud computing” environment or as a “software as a service” (SaaS). For example, at least some of the operations may be performed by a group of computers (as examples of machines including processors), these operations being accessible via a network (e.g., the Internet) and via one or more appropriate interfaces (e.g., application program interfaces (APIs).)

The performance of certain of the operations may be distributed among the one or more processors, not only residing within a single machine, but deployed across a number of machines. In some example embodiments, the one or more processors or processor-implemented modules may be located in a single geographic location (e.g., within a home environment, an office environment, or a server farm). In other example embodiments, the one or more processors or processor-implemented modules may be distributed across a number of geographic locations.

Some portions of this specification are presented in terms of algorithms or symbolic representations of operations on data stored as bits or binary digital signals within a machine memory (e.g., a computer memory). These algorithms or symbolic representations are examples of techniques used by those of ordinary skill in the data processing arts to convey the substance of their work to others skilled in the art. As used herein, an “algorithm” is a self-consistent sequence of operations or similar processing leading to a desired result. In this context, algorithms and operations involve physical manipulation of physical quantities. Typically, but not necessarily, such quantities may take the form of electrical, magnetic, or optical signals capable of being stored, accessed, transferred, combined, compared, or otherwise manipulated by a machine. It is convenient at times, principally for reasons of common usage, to refer to such signals using words such as “data,” “content,” “bits,” “values,” “elements,” “symbols,” “characters,” “terms,” “numbers,” “numerals,” or the like. These words, however, are merely convenient labels and are to be associated with appropriate physical quantities.

Unless specifically stated otherwise, discussions herein using words such as “processing,” “computing,” “calculating,” “determining,” “presenting,” “displaying,” or the like may refer to actions or processes of a machine (e.g., a computer) that manipulates or transforms data represented as physical (e.g., electronic, magnetic, or optical) quantities within one or more memories (e.g., volatile memory, non-volatile memory, or a combination thereof), registers, or other machine components that receive, store, transmit, or display information.

As used herein any reference to “one embodiment” or “an embodiment” means that a particular element, feature, structure, or characteristic described in connection with the embodiment is included in at least one embodiment. The appearances of the phrase “in one embodiment” in various places in the specification are not necessarily all referring to the same embodiment.

Some embodiments may be described using the expression “coupled” and “connected” along with their derivatives. It should be understood that these terms are not intended as synonyms for each other. For example, some embodiments may be described using the term “connected” to indicate that two or more elements are in direct physical or electrical contact with each other. In another example, some embodiments may be described using the term “coupled” to indicate that two or more elements are in direct physical or electrical contact. The term “coupled,” however, may also mean that two or more elements are not in direct contact with each other, but yet still co-operate or interact with each other. The embodiments are not limited in this context.

As used herein, the terms “comprises,” “comprising,” “includes,” “including,” “has,” “having” or any other variation thereof, are intended to cover a non-exclusive inclusion. For example, a process, method, article, or apparatus that comprises a list of elements is not necessarily limited to only those elements but may include other elements not expressly listed or inherent to such process, method, article, or apparatus. Further, unless expressly stated to the contrary, “or” refers to an inclusive or and not to an exclusive or. For example, a condition A or B is satisfied by any one of the following: A is true (or present) and B is false (or not present), A is false (or not present) and B is true (or present), and both A and B are true (or present).

In addition, use of the “a” or “an” are employed to describe elements and components of the embodiments herein. This is done merely for convenience and to give a general sense of the invention. This description should be read to include one or at least one and the singular also includes the plural unless it is obvious that it is meant otherwise.

While particular embodiments and applications have been illustrated and described, it is to be understood that the disclosed embodiments are not limited to the precise construction and components disclosed herein. Various modifications, changes and variations, which will be apparent to those skilled in the art, may be made in the arrangement, operation and details of the method and apparatus disclosed herein without departing from the spirit and scope defined herein.