Moving tables across nodes in an in-memory database instance

The present disclosure involves systems, software, and computer implemented methods for moving a table from a source node to a destination node. In one example, the method includes receiving metadata associated with an in-memory database table stored at a source node. A table container is created responsive to receiving the metadata. The destination node sequentially requests, from the source node, portions of the table, wherein the table is serialized at the source node to provide a serialized sequence of table portions. Sequentially requesting comprises sending a request for a next portion of the table after processing a received portion, which includes receiving a portion of the serialized table, deserializing the received portion, adding the deserialized portion to the created table container, and in response to an end of file indication associated with the received portion, ending the requests and finalizing the table.

TECHNICAL FIELD

The present disclosure relates to computer systems and computer-implemented methods for moving portions of a table in an in-memory database instance from a source node to a destination node.

A directed acyclic graph (“DAG”) is a directed graph with no directed cycles. That is, it is formed by a collection of vertices and directed edges, each edge connecting one vertex to another, such that there is no way to start at some vertex v and follow a sequence of edges that eventually loops back to v again. DAGs can be used to represent a storage hierarchy of a table within a database.

In-memory databases are database management systems that rely on main memory for computer data storage. It is contrasted with database management systems that employ a disk storage mechanism. Main memory databases are faster than disk-optimized databases since the internal optimization algorithms are simpler and execute fewer CPU instructions. Accessing data in memory eliminates seek time when querying the data, which provides faster and more predictable performance than disk. In many cases, data within an in-memory database may be stored on one or more DAGs.

SUMMARY

The present disclosure involves systems, software, and computer-implemented methods for moving a table from a source node to a destination node. In one example, the method includes receiving, at a destination node, table metadata associated with a table of an in-memory database, located at a source node. A table container is created in response to receiving the table metadata. The destination node sequentially requests, from the source node, portions of the table, wherein the table is serialized at the source node to provide a serialized sequence of table portions, and wherein each request includes a sequence number, and wherein sequentially requesting comprises sending a request a next portion of the table including a next sequence number after processing a received portion including a sequence number immediately preceding the current sequence number. Processing the received portion includes receiving a requested portion of the serialized table at the destination node, deserializing the received portion of the serialized table at the destination node, adding the deserialized portion to the created table container in order based on the sequence number associated with the received portion, and in response to determining an end of file indication is associated with the received portion, ending the requesting of portions of the table and finalizing the table based on the received portions.

DETAILED DESCRIPTION

The present disclosure describes a system and methodology for moving a table from one node (i.e., “a source node”) to another (i.e., “a destination node). The full move operation is a three-step process, wherein metadata associated with the table is moved to the new location where a proxy table container is created on the destination node. The data from the table on the source node is then moved to the destination. Upon completion of the move, the table container from the source node can be removed. The present disclosure focuses on a technique and system for the physical move of the table from the source node to the destination node.

A table container is represented in memory as a DAG and can be indexed on a disk using a container directory. Due to the file structure, the table cannot be transferred with simple byte streaming over a network protocol. To achieve serialization for the network transfer, existing functionality can be used in a new manner through enhancement and modification. Specifically, existing file import and export mechanisms can be used to successfully serialize the table persistence to ensure correctness post-move. For in-memory databases, the table may be an extremely large persistence. For that reason, it may be advantageous to avoid materializing a serialized representation on the source node unnecessarily. To do so, and to allow full and correct transmission of the table, a protocol has been developed to allow the import and export mechanism to write into a stream buffer instead of a file, and to transfer the serialized data buffer by buffer from the source node to the destination node over the network.

The DBMS stores some parts of a table in a common page-based persistence container (the “table persistence” or “table container”). Table metadata and, in some instances, virtual files storing data associated with a table, may be stored outside of the table container. In general, these parts are persisted in a single table container, including main storage, delta storage, multi-version concurrency control (MVCC) information, and persisted indexes. The persisted structures of a table are stored in chains of linked pages in the DAG format.

One goal of the table container is to support very big tables that do not fit into standard memory. To obtain maximum performance, in-memory databases may keep as much data in main memory as possible. However, with very big tables, this is not possible, or even desirable from a resource utilization point of view. Large tables can be loaded into memory only partially, and only by loading and unloading parts of the table. The units loaded into memory may need to be big enough to achieve sufficient compression and to enable fast in-memory column operations. However, the parts need to be small enough to fit into memory. In addition, it needs to be ensured that any delta merge operations are also efficient for big tables.

The general design for the table container data transfer between the source node and the destination node is described herein. The data transfer for the table container, which can contain, e.g., delta fragments and optionally paged main fragments, is performed at the first access of the partially moved table (i.e., after the metadata has been moved to the destination node). The data move is accomplished by establishing an internode data stream between a table container export on the source node and a table container import on the destination node. A stream buffer interface is provided to the source and destination nodes, which presents an input and an output stream to the table import and table export providers while internally providing a buffered data transfer over remote procedure calls (RPCs) from source to destination nodes. The table container transfer may be the movement of a table container that represents a portion of a larger table spread across a plurality of nodes (e.g., a partitioned table). In other instances, the table container transfer may be the movement of a single table container holding an entire table, such that the table and table container can be stored on a single node (e.g., an unpartitioned table). Portions of the description may describe either a partitioned or an unpartitioned example. However, it should be understood that the techniques and operations described herein could be used for either type of table.

The destination, or import, context can start via a call to the proxy table container on the destination node. A remote procedure call sent by the destination node can trigger the transfer of the table container. The data transfer will be in the form of serialized chunks of data, which are produced by the export context. The serialized chunks of data are then consumed by a series of remoteRead RPC calls from the destination node to the source node, which in turn fills a stream buffer with the data chunk and provides those data chunks to the import stream. In one implementation, the destination context is single-threaded, using the same thread as the job which accesses the table container. The implementation can be optionally optimized by spawning a parallel job to manage the data transfer from the source node while the main thread consumes the data and performs the import.

The source, or export, context can include a transient structure which is created on the source node at the request of the first remoteRead RPC from the destination node. The transient structure can include an execution thread which hooks to an output stream of the stream buffer interface and performs the export. The export thread can survive, along with the transient structure, after the initial RPC returns until either the data transfer completes or the source table container is unloaded.

The data transfer is driven from the destination node. The destination context requests one chunk of data at a time using a fixed buffer, e.g., of size BUFFERSIZE=10 MB. Each request can be labeled with a sequence number. The source context starts producing chunks at the first request (i.e., sequence 0) and checks that the requested chunk matches with a local sequence number of the produced chunk. In the event the destination context aborts and restarts, the destination context will detect a sequence mismatch and either restart the export from the beginning (e.g., where the sequence number requested is 0), or abort and return an error status (e.g., where the sequence number requested is greater than an expected one). In some instances, it may be possible to restart the export job and skip to a given sequence number.

When the export job writes the last chunk, an end of file (EOF) flag in the export context (or another suitable indication of the end of the file) is set and the job will be complete. When the next chunk request comes in and consumes the final chunk, the export context from the source node can be cleaned. When the last chunk request receives the final chunk, the EOF flag is propagated back to the import context, where the stream is notified of the completed transfer. In the event that the chunk request from the destination node returns with an error (e.g., container not found, node down, etc.), an exception can be thrown to terminate the import and data move operation on the destination node.

The destination-driven design of the solution allows for delayed movement of the data. Specifically, the move operation can be completed from the user's point of view before the data is moved. The data can then actually be moved at a later time, either asynchronously or on demand when table access occurs or is attempted at the destination node.

Turning to the illustrated embodiment,FIG. 1is a block diagram illustrating an example system100for moving portions of a table in an in-memory database instance from a source node to a destination node. As illustrated inFIG. 1, system100is a client-server system capable of executing a distributed and multi-node in-memory database instance102. Specifically, system100includes or is communicably coupled with a client180, a source node105, a destination node138, one or more other nodes169, and network170. Although components are shown individually, in some implementations, functionality of two or more components, systems, or servers may be provided by a single component, system, or server. Similarly, in some implementations, the functionality of one illustrated component, system, or server may be provided by multiple components, systems, servers, or combinations thereof. Conversely, multiple components may be combined into a single component, system, or server, where appropriate.

As used in the present disclosure, the term “computer” is intended to encompass any suitable processing device. For example, source node105may be any computer or processing device such as, for example, a blade server, general-purpose personal computer (PC), Mac®, workstation, UNIX-based workstation, or any other suitable device. Moreover, althoughFIG. 1illustrates a source node105as a single system, source node105can be implemented using two or more systems, as well as computers other than servers, including a server pool. In other words, the present disclosure contemplates computers other than general-purpose computers, as well as computers without conventional operating systems. Further, illustrated source node105, destination node138, other nodes169, and client180may each be adapted to execute any operating system, including Linux, UNIX, Windows, Mac OS®, Java™, Android™, or iOS. According to one implementation, the illustrated systems may also include or be communicably coupled with a communication server, an e-mail server, a web server, a caching server, a streaming data server, and/or other suitable server or computer.

In general, the in-memory database instance102represents a plurality of nodes in which fragments, or portions, of a large table may be stored. The in-memory database instance102represents the collection of nodes in which a particular in-memory database or set of databases may be stored, and may refer to one or a plurality of systems, servers, computers, and devices, either physical or virtual.

As illustrated, the in-memory database instance102includes a plurality of nodes, including the source node105, destination node138, and optionally, one or more other nodes169. In the illustrated example, a table stored at the source node105(i.e., table129) is to be moved to the destination node138(i.e., table162). In the in-memory database instance102, each node may be a completely separate computer or system (whether physically or virtually), with each system holding a portion of the in-memory database to provide significant scaling capabilities. Each node in the database can comprise and/or store a portion of an overall table spread across a plurality of nodes. Alternatively, in an unpartitioned example, a particular node can store an entire table without the table being stored on additional nodes. To move a particular table from one node to another, a move command is needed to perform the necessary shifting of the table, e.g., from the source node105to the destination node138. Such a command may be initiated, for example, by a client application182at client180, as well as by a particular application executing on or associated with one of the nodes.

In the present example, the source node105is the original location of a particular table (i.e., table129), which is to be moved to the destination node138via the processes described herein. The source and destinations nodes105,138may include similar or different components. As illustrated, the source node105includes an interface108, a processor111, a database management system (DBMS)114(which includes serializer117and stream buffer120), and memory126. In general, the source and destination nodes105,138are simplified representations of one or more systems and/or servers that provide the described functionality, and is not meant to be limiting, but rather an example of the systems possible.

The interface108is used by the source node105for communicating with other systems in a distributed environment—including within the environment100—connected to the network170, e.g., other nodes within the in-memory database instance102such as the destination node138or the one or more other nodes169, client180, and other systems communicably coupled to the network170. Generally, the interface108comprises logic encoded in software and/or hardware in a suitable combination and operable to communicate with the network170. More specifically, the interface108may comprise software supporting one or more communication protocols associated with communications such that the network170or interface's hardware is operable to communicate physical signals within and outside of the illustrated environment100.

Network170facilitates wireless or wireline communications between the components of the environment100(i.e., between the various nodes, between particular nodes and client180, and among others), as well as with any other local or remote computer, such as additional clients, servers, or other devices communicably coupled to network170, including those not illustrated inFIG. 1. In the illustrated environment, the network170is depicted as a single network, but may be comprised of more than one network without departing from the scope of this disclosure, so long as at least a portion of the network170may facilitate communications between senders and recipients. In some instances, one or more of the illustrated components may be included within network170as one or more cloud-based services or operations. The network170may be all or a portion of an enterprise or secured network, while in another instance, at least a portion of the network170may represent a connection to the Internet. In some instances, a portion of the network170may be a virtual private network (VPN). Further, all or a portion of the network170can comprise either a wireline or wireless link. Example wireless links may include 802.11a/b/g/n, 802.20, WiMax, LTE, and/or any other appropriate wireless link. In other words, the network170encompasses any internal or external network, networks, sub-network, or combination thereof operable to facilitate communications between various computing components inside and outside the illustrated environment100. The network170may communicate, for example, Internet Protocol (IP) packets, Frame Relay frames, Asynchronous Transfer Mode (ATM) cells, voice, video, data, and other suitable information between network addresses. The network170may also include one or more local area networks (LANs), radio access networks (RANs), metropolitan area networks (MANs), wide area networks (WANs), all or a portion of the Internet, and/or any other communication system or systems at one or more locations.

As illustrated inFIG. 1, the source node105includes a processor111. Although illustrated as a single processor111inFIG. 1, two or more processors may be used according to particular needs, desires, or particular implementations of the environment100. Each processor111may be a central processing unit (CPU), an application-specific integrated circuit (ASIC), a field-programmable gate array (FPGA), or another suitable component. Generally, the processor111executes instructions and manipulates data to perform the operations of the source node105. Specifically, the processor111executes the algorithms and operations described in the illustrated figures, including the operations performing the functionality associated with the source node105generally, as well as the various software modules (e.g., the DBMS114, etc.), including the functionality for sending communications to and receiving transmissions from client180and other nodes138,169within the in-memory database instance102.

The illustrated source node105includes memory126. The memory126may include any memory or database module and may take the form of volatile or non-volatile memory including, without limitation, magnetic media, optical media, random access memory (RAM), read-only memory (ROM), removable media, or any other suitable local or remote memory component. The memory126may store various objects or data, including databases, financial data, user information, administrative settings, password information, caches, applications, backup data, repositories storing business and/or dynamic information, and any other appropriate information including any parameters, variables, algorithms, instructions, rules, constraints, or references thereto associated with the purposes of the source node105. Additionally, the memory126may store any other appropriate data, such as VPN applications, firmware logs and policies, firewall policies, a security or access log, print or other reporting files, as well as others. For example, memory126can store portions of the in-memory database instance102within the table129and its table container132and table metadata135. The table container132, as described above, can in some instances store the data of a fragment of a larger table stored across multiple nodes in the multi-nodal in-memory database instance102or, alternatively, can store data defining an individual table that is stored within a single node. The table container132, and the other portions of the in-memory database instance102, can be stored in main memory (e.g., RAM) for computer data storage. While the table container132is stored in main memory, some or a portion of the other information stored in memory126may be stored elsewhere. In addition to the table container132, memory126includes, within the table129, a set of table metadata135that describes the table129stored in memory126. Table container132represents a data storage abstraction provided by the persistence layer and is implemented as collections of pages, although this can be hidden from the caller of the container interface. Containers may be used internally by the persistence layer, where appropriate.

As illustrated, the source node105includes the DBMS114. The DBMS114is a software application that can interact with users, other applications, and the database itself to retrieve, update, delete, and analyze data. DBMS114, in particular, is an application specifically designed to manage and interact with an in-memory database such as the table container132stored at the source node105. For example, the DBMS114may be able to interpret standard-based requests, such as Structured Query Language (SQL) commands, and perform the actions associated therewith upon the databases and tables with which it is associated (e.g., the table container132). The DBMS114may be associated with one or more other applications, or may be a standalone application. In the illustrated example, the DBMS114may assist in moving the table container132from the source node105to the destination node138in response to a SQL move command. The DBMS114includes serializer117and stream buffer module120, each providing additional functionality to the DBMS114.

Regardless of the particular implementation, “software” includes computer-readable instructions, firmware, wired and/or programmed hardware, or any combination thereof on a tangible medium (transitory or non-transitory, as appropriate) operable when executed to perform at least the processes and operations described herein. In fact, each software component may be fully or partially written or described in any appropriate computer language including C, C++, JavaScript, Java™, Visual Basic, assembler, Perl®, any suitable version of 4GL, as well as others.

The serializer117of the DBMS114performs operations to serialize the contents of the table container132(stored as a DAG). By serializing the table container132, the move operations can be performed in data chunks over the network. Serializer117can allow complex data such as querysets, model instances, and DAGs, among others, to be converted to datasets that can then be easily rendered into JSON, XML, or other content types. Serializer117can also provide deserialization, allowing serialized data to be converted back into complex types, after first validating the incoming data.

The output of serializing the data of the table container132is then fed into the stream buffer module120, which can also be a part of the DBMS114. The stream buffer module120provides a generic buffer interface that allows the serialized data from the table container132to be consumed by another node (using that node's remote procedure call156) over the network. In the described examples, the stream buffer module120may be able to accept a fixed chunk of serialized table data. In response to a request from the destination node138(and in particular, its remote procedure call156), the portion of the serialized in the stream buffer can be consumed and provided to the destination node. In one example, the stream buffer module120may have a fixed size of 10 bytes. Therefore, a serialized table may be fed into the stream buffer12010 bytes at a time. Each addition to the stream buffer120may be assigned a sequence number to ensure that the data is received in an ordered fashion, such that the serializer150at the destination node138can deserialize the received data chunks of the table165in the proper order.

As illustrated, a series of chunks136, or portions of the data from table container132, are sent from the source node105to the destination node138. The process describing how the chucks136are transmitted from the source node105to the destination node138are described inFIGS. 2-4.

As described, the destination node138may be similar to or different in design to the source node105. In the illustrated example, the destination node138includes similar aspects and components as the source node105. The destination node138includes an interface141, a processor144, and a DBMS147(which includes serializer150, stream buffer153, and a remote procedure call (RPC)156). The interface141, processor144, and DBMS147may be similar to or different from those described in the source node105. The destination node138includes the RPC156, where the RPC156is used to obtain the chunks136of the data from the table container132. The RPC156can access the stream buffer120of the source node105and retrieve or otherwise obtain the chunks136of the data within the table container132. Those chunks136are placed into the stream buffer153at the destination node138, where those chunks136are then provided to the serializer150for deserialization. Serializer150may be similar to serializer117, and allows the serialized portions of the table data to be deserialized. The deserialized chunks136are then combined to form the table container165, which is stored in memory159(which may be similar to or different than memory126) in table container162along with table metadata168. The table metadata168can be identical to, or include the information from, the table metadata135, and can be provided to the destination node138in response to the move instruction. When the table metadata135is received, a proxy, or empty, table container162can be generated and prepared for the moved table data. In some instances, the receipt of the table metadata168and the generation of the table container162can occur before the movement of the table data and, in some instances, may trigger the RPC156used to initiate the move operations.

The RPC156is an inter-process communication that allows a computer program, here the DBMS147, to cause a subroutine or procedure to execute in another address space (i.e., on the source node105) without requiring explicitly coding for the details of the remote interaction. In the present example, the RPC156provides instructions to read table data from the stream buffer120and return that table data to the destination node138as chunk136. Once the chunks136are provided to the stream buffer153and the serializer150(for deserialization), the table container165is recreated at the destination node138. Each request can result in a new RPC instance, as each RPC itself only handles a single chunk136. The series of RPC calls are controlled by the import context and stream buffer on the destination node138. The series of RPC instances156can continue requesting and receiving chunks136of the data within the table container132until an end of file indication is received. In some instances, each chunk136may include a payload, a sequence number, and a field for an end of file indication of flag. If the end of file indication is included, no further RPC instances156requesting new chunks of the table container132and its table data are sent. In some instances, the series of RPC instances156can request for portions of the table container132in order, where each RPC instance156identifies a particular sequence number for retrieval. By requesting the table in a specific order, the serialization and deserialization operations can ensure that the table container132and its table data is recreated at the destination node138in the correct order. This is especially important where the data in the table container132is stored, for instance, as a DAG.

While in the present illustration the element labeled105is a source node while the element labeled138is a destination node138, the roles of the nodes, as well as the other nodes169, may be changed in alternative operations. In other words, a request to move table162to the source node105may be received, such that a reverse process is performed. The various nodes are each capable of performing the move operations in either direction and may include any necessary components for doing so.

Other nodes169comprise one or more nodes within the in-memory database instance102where further portions of the multi-nodal database are stored. The other nodes169may be similar to or different than either the source or destination nodes105,138. In further examples, a move command may be performed moving tables from either the source node105or destination node138to one of the other nodes169, as well as moving tables from a particular other node169to one of the source node105or the destination node138.

In some instances, a master node within the in-memory database instance102may contain information identifying where particular tables and portions of tables exist within the in-memory database instance102. For instance, requests to move a particular table may first be sent to the master node, which uses its stored information to determine which particular nodes hold the relevant table and to where the table is to be moved. The master node can then translate a request to move a particular table to the proper form to cause the source node105to move the table to the destination node138. In some instances, either the source node105or the destination node138can be the master node.

Client180may be any computing device operable to connect to or communicate with the nodes of the in-memory database instance102via network170, as well as the with the network170itself, using a wireline or wireless connection, and can include a desktop computer, a mobile device, a tablet, a server, or any other suitable computer device. In general, client180comprises an electronic computer device operable to receive, transmit, process, and store any appropriate data associated with the environment100ofFIG. 1. In the illustrated example, client180may include a client application182capable of issuing requests that may trigger or cause movement of a table from the source node105to the destination node138, as well as to request information from the tables included within the in-memory database instance102. Prompts and other information associated with the move and the in-memory database instance102may be presented at the client180, such as at a graphical user interface (GUI).

While portions of the software elements illustrated inFIG. 1are shown as individual modules that implement the various features and functionality through various objects, methods, or other processes, the software may instead include a number of sub-modules, components, libraries, and such, as appropriate. Conversely, the features and functionality of various components can be combined into single components as appropriate.

FIG. 2is an illustration of example operations performed to move portions of a table in an in-memory database instance from a source node to a destination node. The top ofFIG. 2illustrates operations at a source node205, while the bottom ofFIG. 2illustrates operations at the destination node210.FIG. 2is meant as an example illustration. Alternative implementations, additional or fewer operations, and modifications to the illustration are understood to be within the scope of the present disclosure.

Initially, the source node includes a table container220storing table data represented as a directed acyclic graph (DAG). The table container220may be similar to table container132described above. Because of this, a direct transfer of the data within the table container220from the source node205to the destination node210is not possible. Initially, a set of table metadata223(stored apart from the table container220itself and describing the table data) is provided to the destination node210. As a result, the table container260is created. As illustrated, portions of the table data are provided to a serializer225, where the serializer225outputs a serialized version of the table data included within the table container220. Portions of the serialized table data are provided to a stream buffer230, where the stream buffer is of a fixed size (e.g., 10 bytes). The stream buffer230(or serializer225) can assign, to each portion of the serialized table data, a particular sequence number beginning, for example, at 0. As portions are consumed, new portions are added to the stream buffer230, each with the next sequence number in order. When those sequences are deserialized later and combined in the specified order, the move of table container220can be completed.

As illustrated, a series of remote procedure calls240from the destination node210are provided to request for the movement of the table including table container220to the destination node210, as well as to perform the consumption of the serialized portions within the stream buffer230. The remote procedure calls240can access the stream buffer230and consume the serialized portions (e.g., data buffer235). The consumed data can be returned to the destination node210(e.g., data buffer245) and placed into the input stream buffer250. As the stream buffer250is filled, the corresponding data is sent to the deserializer255, where the deserializer255performs operations to return the table data into the DAG format of table container260.

In an example operation of the illustration actions, the first remote procedure call instance240can request sequence 0 of the serialized table data from the table container220in response to a request to move the table to destination node210. In some instances, the request may be associated with the receipt of table metadata223associated with the table and its table container220at the destination node210, along with information identifying the move request. In response to the request from the first remote procedure call instance240, the source node205may begin serializing the table data and placing segments into the stream buffer230. When sequence 0 is placed into the stream buffer230, a first RPC procedure call240accesses the stream buffer230and returns the data buffer235associated with sequence 0 to the destination node210. Sequence 0 can be placed into the stream buffer250, which is then sent to the deserializer255, where the deserialized data is reformed into the table container260. The process can continue such that the next sequence number is requested by a new remote procedure call instance240, and the remote procedure call instance240can access and return data chunk 1, similarly providing data chunk 1 to the stream buffer250and deserializer255. Each RPC consumes a single sequence, and the process continues until all portions of table data within the table container220are provided and consumed by the series of remote procedure calls240. In some instances, when the final portion of the table data is serialized, an end of file (EOF) flag or other indication can be set when the final data chunk is placed into the stream buffer230. When the last remote procedure call instance240identifies the EOF flag, the remote procedure call instance240will return the data to the destination node210and ceases requesting additional sequences with another instance of the remote procedure call240. After adding the final sequence to the table container260, the destination node210can finalize the table container260and make the completed table available, such as by removing any locks applied during the move.

In some instances, the source node205can perform cleanup activities to remove any remnants of table container220, including erasing or destroying any table information, data, or metadata associated with table container220. Additionally, confirmation of the move may be reported to a master node, as appropriate, to ensure records on the new location of table container260and its table data are maintained and available.

FIG. 3is a flowchart of an example operation300from the perspective of a destination node related to moving portions of a table in an in-memory database instance from a source node to a destination node. For clarity of presentation, the description that follows generally describes method300in the context of the system100illustrated inFIG. 1. However, it will be understood that method300may be performed, for example, by any other suitable system, environment, software, and hardware, or a combination of systems, environments, software, and hardware as appropriate.

At305, table metadata from the source node is received at the destination node. The table metadata can be sent by the source node in response to an initial move table request. Upon receiving the table metadata, the destination node creates a proxy, or empty, table container on the destination node at310. The table container will be associated with the table metadata, and can be used to contain the incoming table.

At315, the destination node creates an import context and initiates the data import from the source node. The import context can include an instantiation of one or more remote procedure calls to facilitate the move, as well as the preparation of a stream buffer at the destination node. Once prepared, the remote procedure call (or a new instance thereof) sends a request to the source node for the next portion of the table data to be moved at320. The remote procedure call can correspond with the stream buffer of the source node. In some instances, the first request from the remote procedure call (or an instance thereof) of the destination node may trigger the serialization and stream buffer operations of the source node. The request from the remote procedure call is associated with a particular sequence number of the serialized table data, as the serialized table data is split into multiple pieces and added to the source node's stream buffer. Normally, unless the table data within the table container is small enough to fit into a small amount of data buffers, the stream buffer will only contain a small portion of the table data from the table container to avoid materializing the entire serialized representation in memory. In many instances, there is a limit to the size of the data buffer(s) in the stream buffer. In the present instances, a single data buffer associated with a sequence number may be included within the stream buffer. However, in alternative implementations, the stream buffer could be expanded to employ multiple data buffers. The remote procedure call can initially request a first sequence (e.g., sequence 0).

At325, a determination is made as to whether the remote procedure call identifies an error. An error may occur when the sequence number associated with the portion of the table data in the source node's stream buffer does not match the requested sequence number. Other errors may also be possible. When an error is detected, method300moves to330, where an error is returned and the import process ends. If, however, no error is determined, method300continues at335, where the remote procedure call can consume the portion of the table data, or table data chunk, from an RFC data buffer. The remote procedure call can return the consumed portion of the table data into the import stream buffer at the destination node. The import stream buffer can be the same size as the export stream buffer, thereby allowing the full portion of the table data consumed from the export stream buffer at335to be placed into the import stream buffer. Alternatively, the import stream buffer may be bigger than the export stream buffer to allow additional chunks to be delivered to the destination node, such as when delays in deserializing the imported chunks slows the destination node.

At340, the contents of the import stream buffer can be read into the serializer at the destination node, and at345, the serializer can perform the deserialization operations for the particular sequence. The deserialization operations allow the previously serialized portion of the table data to be translated back into its original state—here, into the DAG format. Also at345, the deserialized portions of the table data are added to the table container. As each of the deserialized portions are added in the proper sequence, the full table can be moved to the destination node.

After, during, or concurrently to the operations of335through345, method300determines whether an end of file (EOF) indication is included in the current serialized sequence being processed at350. If not, method300returns to320, where a new request is sent from the remote procedure call (or an instance thereof) to obtain the next sequence of the serialized table data from the export stream buffer of the source node. If, however, the EOF indication is identified, method300moves to355, where the series of remote procedure calls ends the requests and the table is finalized within the table container at the destination node.

FIG. 4is a flowchart of an example operation400from the perspective of a source node related to moving portions of a table in an in-memory database instance from a source node to a destination node. For clarity of presentation, the description that follows generally describes method400in the context of the system100illustrated inFIG. 1. However, it will be understood that method400may be performed, for example, by any other suitable system, environment, software, and hardware, or a combination of systems, environments, software, and hardware as appropriate.

At405, a request to move a table currently at the source node to the destination node is received. In some instances, the request may be received from a separate master node other than the source node itself within an in-memory database instance. In response to receiving the request, at410, the source node can send a set of table metadata associated with the to-be moved table to the destination node associated with the request. The table metadata can be sent in any appropriate format and using any appropriate protocol.

At415, a source node export context is created to perform the operations needed to move the table from the source node to the destination node. The export context may include the creation of an export stream buffer. In some instances, the source node export context is not created until a first remote procedure call is received. In those instances, the creation of the source export context may be delayed until after the first request, such as at435and440below.

At420, a request from a remote procedure call of the destination node is received, where the request identifies a particular sequence number of a serialized set of table data from the table container. At425, a determination is made as to whether the request was associated with an expected sequence number. For example, if the request was the first request and included a request for the first sequence (e.g., sequence 0), then the expected sequence would be received. If the prior request was for sequence 2 and the current request was for sequence 8, then the expected sequence would not be received. If the determination is made that the correct sequence number is requested, method400continues at435. If, however, the determination is made that an incorrect sequence number is requested, then method400moves to430where an error is returned.

At435, a determination is made as to whether a source export context exists. As described above with regard to415, the source export context may not be created until the first request (i.e., remote procedure call) is received. If the creation of the source export context is delayed until the first request is received, then upon receiving the first request, a source export context is created at440. During portions of the loop where the source export context exists, method400continues at445.

At445, at least a portion of the table data (which is stored as a DAG) from the table container is serialized. In some instances, the table data may be serialized at or in response to the source export context being created. In other instances, such as is illustrated here, only a portion of the table data may be serialized in response to the incoming request from the remote procedure call. In some instances, the entire table data within the table container may be serialized at once when the entire serialized representation fits into the fixed number of data buffers allocated by the export stream buffer. At450, the serialized portion of the table data is sent to the fixed-size import stream buffer until the stream buffer is full. Once the stream buffer is full, the remote procedure call from the destination node consumes the data.

At455, a determination is made as to whether the table serialization is complete. If the current serialized portion of the table data to be provided to the stream buffer is the last from the table container, then the table serialization is considered complete. If additional portions remain, then method400returns to420and awaits a request for the next sequence of the serialized table data. If, however, the table serialization is complete, at460the current data chunk written into the stream buffer can have its end of file flag or other indication set to provide the destination node and its remote procedure call with the information that the complete table has been provided.

At465, once the table serialization is complete and the table move is completed, the source node can perform cleanup operations to remove any remaining portions or related content to the table. In some instances, the cleanup operations may not be executed until a confirmation of completion is received from the destination node. Waiting to perform cleanup of the source import context can be beneficial in cases where a transient network error may cause the failure of an RPC instance, after which the destination export context may re-try the RPC to get the currently requested sequence. The source cleanup may include two parts: 1) cleanup of the export context which happens as described here and 2) removal of the persistent table container from the source node. In case of a crash, the data will be maintained to guarantee that the data has been written to disk on the destination node or is still accessible on disk at the source node; otherwise, data loss can occur. By maintaining the data until the save point is reached, data recovery in the event of a crash is ensured. The removal of the persistent table container may only happen after a save point has occurred on the destination node ensuring that the table container has been written to disk and ensuring it is persisted.

The preceding figures and accompanying description illustrate example systems, processes, and computer-implementable techniques. While the illustrated systems and processes contemplate using, implementing, or executing any suitable technique for performing these and other tasks, it will be understood that these systems and processes are for illustration purposes only and that the described or similar techniques may be performed at any appropriate time, including concurrently, individually, or in combination, or performed by alternative components or systems. In addition, many of the operations in these processes may take place simultaneously, concurrently, and/or in different orders than as shown. Moreover, the illustrated systems may use processes with additional operations, fewer operations, and/or different operations, so long as the methods remain appropriate.