Patent Description:
The approaches described in this section could be pursued, but are not necessarily approaches that have been previously conceived or pursued. Therefore, unless otherwise indicated herein, the approaches described in this section are not prior art to the claims in this application and are not admitted to be prior art by inclusion in this section.

Data communication networks may comprise one or more computing nodes. Each node may receive and process unbounded, streaming data according to a particular schema. A schema may define the content, and/or the format of the content, in a data stream. Each node may send results to one or more downstream nodes, persistent storages, and/or some other receiving device, according to a schema.

In a modern data-driven system, the schema or format of a data stream may change at will. For example, a schema for a network data stream may include frequent status reports for each mission critical device on the network, which includes the device's Internet Protocol address and a time stamp and may change to include the power consumption, CPU usage, and available memory on each device.

A change in a data stream's schema may cause may cause a node to stop. However, some scenarios require that a node remain processing, in which cases the node is not allowed to be suspended or reconfigured to take advantage of a schema change.

<NPL> describes, according to its abstract, the specification of static metadata for streams in a model called Stream Schema. Stream Schema can be used to validate the consistency of streams. By explicitly modeling stream constraints, stream queries can be simplified by removing predicates or subqueries that check for consistency to enhance programmability of stream processing systems. This document also describes a set of semantic query optimization strategies that both permit compile-time checking of queries (for example, to detect empty queries) and new runtime processing options, options that would not have been possible without a Stream Schema specification.

<CIT> describes, according to its abstract, a method, system, and computer program product for merging monitoring data streams which includes: receiving a first monitoring data stream relating to a server containing first records, each first record has a first transaction identifier relating to a first transaction and first monitoring data, relating to a performance of executing the first transaction; receiving a second monitoring data stream relating to a client containing second records, each second record has a list of second transaction identifiers relating to one or more second transactions and second monitoring data, relating to a performance of executing the second transactions; buffering at least parts of the first and the second monitoring data streams; and merging the buffered first and second monitoring data streams by computing an outer-join with a join-condition that the first records have the first transaction identifier that occurs in the list of second transaction identifiers to provide a merged data stream.

While each of the drawing figures illustrates a particular embodiment for purposes of illustrating a clear example, other embodiments may omit, add to, reorder, and/or modify any of the elements shown in the drawing figures. For purposes of illustrating clear examples, one or more figures may be described with reference to one or more other figures, but using the particular arrangement illustrated in the one or more other figures is not required in other embodiments.

Techniques for distributing and processing independent data streams over one or more networks are described. In the following description, for the purposes of explanation, numerous specific details are set forth in order to provide a thorough understanding of the present disclosure. It will be apparent, however, to one skilled in the art that the present disclosure may be practiced without these specific details. In other instances, well-known structures and devices are shown in block diagram form in order to avoid unnecessarily obscuring the present disclosure.

Embodiments are described herein according to the following outline:.

The invention to which the present European patent relates is defined in the appended claims.

In an embodiment, a computer system comprises a first node computer comprising a first processor and a first memory configured to process one or more first type of records based on a first operation; a second node computer comprising a second processor and a second memory configured to process one or more second type of records based on a second operation; a routing computer comprising a processor, a memory, and a routing module configured to: receive a first record comprising one or more first fields from a first source computer; in response to determining that the first record is a first type and the first node computer is associated with the first type, send the first record to the first node computer to be processed; receive a second record comprising the one or more first fields and one or more second fields from a second source computer; in response to determining that the second record is a second type and the second node computer is associated with the second type, sending the second record to the second node computer to be processed; in response to determining that the second type is a first subtype of the first type, sending the second record to the first node computer to be processed, without suspending and reconfiguring the first node computer.

In an embodiment, a method comprises receiving a first record comprising one or more first fields from a first source computer; in response to determining that the first record is a first type and a first node is associated with the first type, sending the first record to the first node to be processed; receiving a second record comprising the one or more first fields and one or more second fields from a second source computer; in response to determining that the second record is a second type and a second node is associated with the second type, sending the second record to the second node; in response to determining that the second type is a first subtype of the first type, sending the second record to the first node to be processed, without suspending the first node.

In an embodiment, the method comprises in response to determining that the first record is the first type and the first node is associated with the first type, sending a first signature with the first record to the first node; in response to determining that the second record is the second type and the second node is associated with the second type, sending a second signature with the second record to the first node; in response to determining that the second type is the first subtype of the first type, sending the first signature with the second record to the first node.

In an embodiment, the method comprises organizing the one or more first fields in the first record according to one or more first indices; organizing the one or more first fields in the second record according to the one or more first indices; organizing the one or more second fields in the second record according to one or more second indices, wherein each index in the one or more first indices references a memory address that is before the memory address referenced by each index in the one or more second indices.

In an embodiment, the method comprises receiving, from a third source computer, a third record comprising the one or more first fields, the one or more second fields, and one or more third fields; in response to determining that the third record is a third type and a third node is associated with the third type, sending the third record to the third node to be processed; in response to determining that the third type is a second subtype of the second type, sending the third record to the second node to be processed without suspending and reconfiguring the second node; in response to determining that the second type is the first subtype of the first type, sending the third record to the first node to be processed without suspending and reconfiguring the first node.

Suspending a node may cause many problems. Suspending may include halting a node and/or otherwise causing a node to stop processing one or more input data streams. For example, a node may process time-critical data, such as networking, power, or rocket telemetry data. If the node is suspended, a downstream node or device may be unable to perform its functions properly or may derive misleading data from the lack of data received for the suspended node. For example, the downstream node or device may determine that a network is down or a rocket has lost contact, however the data not processed by the suspended node may indicate that a network is excellent connectivity with one or more other networks or that a rocket is performing within mission parameters.

Suspending a node may cause massive reserve resources dedicated to the node or data loss. For example, a node may process a data stream that continuously delivers a large amount of data. If the node is suspended, then a large amount of data may begin to build up, which may need to be stored in a large persistent data store. Even data may still be lost even if unprocessed data is stored in the large persistent data store. For example, if the node is offline for too long, then the data store fill to capacity. Thus, either new data may not be stored, or the older, unprocessed data may be deleted to make room for the newly received data. Furthermore, if the bandwidth of the data received is greater than the bandwidth of the persistent storage, then some data may be lost or overwritten while waiting in a volatile buffer waiting to be written to persistent storage.

Suspending a node may cause one or more data sources to fail. For example, if a data source receives an exception error because a receiving node is suspended, then the data source may exit. Thus, even with the suspended node resumes processing the data source may have terminated.

In response to schema changes, a node may be suspended, reconfigured, and resumed. Continuing with the previous example, if the data consumer administrating the node wishes to process and/or use the new data in the data stream (the power consumption, CPU usage, and available memory on each device), then the administrator may suspend the node, reconfigure the node to make use of the new data, and resume the node. The data sent while the node is offline may be lost or, if possible, stored in a persistent storage when the node comes back online.

After a node resumes processing, the node may not immediately begin processing the most recent data. Continuing with the previous example, when the reconfigured node resumes, the node may process the data stored in persistent storage first, meanwhile newly received data may continue to be stored in persistent storage or lost. For example, if the persistent storage's bandwidth is insufficient to support the reads requested by the node and the writes from the data stream, then the data in the data stream may be lost while the data from persistent storage is processed by the node.

A change in a data stream's schema may cause a node to be suspended, which in turn may cause many problems, as discussed above, and may be prone to error. Furthermore, some scenarios require that a node remain processing, in which cases the node is not allowed to be suspended or reconfigured to take advantage of a schema change. Thus, gracefully and efficiently supporting schema changes without downtime is extremely valuable.

One or more systems and methods are discussed herein to process input data streams that conform to changing schemas without suspending and/or reconfiguring running nodes. Suspending a node may mean stopping a node from processing and/or executing one or more continuous queries that are based on a first type of record. A continuous query may comprise one or more continuous queries, sub-queries, operations defined in a continuous query, and/or sub-operations defined in a continuous query. Reconfiguring a node may mean suspending a node and/or causing the node to execute and/or process one or more new continuous queries based on a second, different type of record that is a subtype of the first type of record. Additionally or alternatively, reconfiguring a node may mean causing a node to execute and/or process one or more new continuous queries that perform a super set of the functionality of the one or more first continuous queries.

Data streams may comprise one or more records sent over time. A record in a data stream may be organized according to a schema. Records that are organized according to a first schema may be referred to as a "first type" or a "first type of record". Records that are organized according to a second schema may be referred to as a "second type" or "second type of record".

One or more systems and methods are discussed herein for a first node, configured to process a first type of record, to also process a second type of record without suspending and/or reconfiguring the first node. For example, a record that is a second type of record may be cast as the first type of record and processed by the first node without suspending and/or reconfiguring the first node. Accordingly, one or more systems and methods are discussed herein to cast a second type of record as a first type of record, and process the second type of record as if the record was originally received as a first type of record.

A record may comprise one or more fields. A record may also comprise an express signature. A "bare" record does not include an express signature. A bare record may, but need not, include an implied signature. A signature may include one or more values and/or data structures that identify the record a particular type of record. One or more of the systems and/or methods discussed herein may append and/or prepend a signature to a record, such that the record is no longer a bare record.

<FIG> illustrates a plurality of records that are one or more types in an example embodiment. In <FIG>, record <NUM> is an example first type of record and comprises signature <NUM> and fields 114A. Signature <NUM> may identify record <NUM> as a first type of record. Fields 114A may comprise one or more values, key-value pairs, lists of values and/or key-value pairs, hierarchical data structures, pointers, and/or any other data structures.

Record <NUM> is an example second type of record and comprises signature <NUM>, fields 114B, and fields 124A. Signature <NUM> may identify record <NUM> as a second type of record. Fields 114B in record <NUM> has the same format and/or data structure(s) as fields 114A in record <NUM>, but the values may be different. Fields 124A may comprise one or more different and/or additional values, key-value pairs, lists of values and/or key-value pairs, hierarchical data structures, pointers, and/or any other data structures that are not included in fields 114B.

In the example records illustrated in <FIG>, each record includes an "express" signature comprising value and/or data structure. For example, record <NUM> includes signature <NUM>, and record <NUM> includes signature <NUM>. However, in an embodiment, a signature may be implied by one or more names, values, and/or data structures of one or more fields in the record. For example, any record that includes a field named "address" may be determined to be an instance of a first type, and any record that includes a field named "distance" may be determined to be an instance of a second type. A record that includes both a field name "address" and a field name "distance" may be determined to be an instance of both the first type and the second type.

A second type of record may be referred to herein as a "subtype" of a first type of record if the fields in the second type of record comprise a superset of the required fields in the first type of record. The first type may be referred to herein as a "super-type" and/or a "subset" of the second type. For purposes of illustrating a clear example, assume that the one or more fields in fields 114A are required for a first type of record, and the one or more fields in fields 114B and fields 124A are required for any instance of a second type of record. In the current example, a record that is an instance of the second type also includes the required fields to be an instance of the first type. Thus, the second type is a "subtype" or "superset" of the first type; and, the first type is a "super-type" or "subset" of the second type.

A field may be a required field if it is used by one or more nodes that process the first type of records. Additionally or alternatively, a field may be a required field if the field is designated as required in a schema. For example, a first schema, which defines the first type of record, may include a list of the names of the fields that are required. Additionally or alternatively, the first schema may identify one or more fields in a first data structure as required and one or more fields in a second data structure as optional.

A record may have more than one subtype and/or super-type. For example, record <NUM> is an example of a third type of record. Record <NUM> comprises signature <NUM>, fields 114C, fields 124B, and fields <NUM>. Signature <NUM> may identify record <NUM> as a third type of record and/or as a record that may be processed by one or more third nodes. Fields 114C in record <NUM> has the same format and/or data structure(s) as fields 114A in record <NUM>, but the values may be different. Fields 124B in record <NUM> has the same format and/or data structure(s) as fields 124A in record <NUM>, but the values may be different. In the embodiment illustrated in <FIG>, the third type of record is a subtype of the first type of record, because the third type of record has the one or more required fields in the first type (fields 114A in record <NUM> and fields 114C in record <NUM>). The third type of record is a subtype of the second type of record, because the third type of record has the one or more required fields in the second type (fields 124A in record <NUM> and fields 124B in record <NUM>).

The signature and fields of each record in <FIG> are illustrated as if contiguous in memory. However, the signature and/or fields of each record need not be contiguous. For example, signature <NUM>, fields 114B, and fields 124A in record <NUM> may, but need not, be stored contiguously in memory. Furthermore, each field in fields 114A and/or fields 114B may, but need not, be stored contiguously in memory.

Records that are a subtype of another type may be cast as a super-type by changing the signature, fields, and/or data structures in the records. For example, record <NUM> is a copy of record <NUM>, which is a second type of record, cast as a super-type: the first type. Record <NUM> comprises signature <NUM>, fields 114B, and fields 124A. Just as in record <NUM>, signature <NUM> in record <NUM> may include one or more values and/or data structures that identify record <NUM> as a first type and/or as a record to be processed by one or more first nodes associated with signature <NUM> and/or the first type. However, the one or more first nodes may disregard fields 124A in record <NUM>.

One or more fields that are not in a super-type may be stored higher memory address space. For example, in <FIG>, memory addresses grow from left to right. Thus, in record <NUM>, a first index that references fields 114B is less than a second index that references fields 124A. A first node that processes the first type of record may use the same index to reference fields 114A in record <NUM> and fields 114B in record <NUM>, because fields 124A are assigned a higher memory address. Thus, the first node may process record <NUM> as if record <NUM> was originally a first type of record.

Common fields between a subtype and a super-type may be stored in the same order. For example, fields 114C and fields 124B in record <NUM> are in the same order as fields 114B and fields 124A in record <NUM>. A node that processes the second type of records may use the same indices to reference fields 114B and fields 124A in record <NUM> as fields 114C and fields 124B in record <NUM>. Furthermore, a node that processes the first type of records may use the same index to reference fields 114A in record <NUM>, fields 114B in record <NUM>, fields 114B in record <NUM>, and fields 114C in record <NUM>.

<FIG> illustrates a system for processing different types and subtypes of streaming input data from a plurality of sources without suspending one or more nodes in an example embodiment. In <FIG>, system <NUM> includes node computer cluster <NUM>, routing computer <NUM>, record storage device <NUM>, source computer <NUM>, source computer <NUM>, source computer <NUM>, and client computer <NUM> distributed across a plurality of interconnected networks. Node cluster <NUM> includes source node computer <NUM>, query node computer <NUM>, source node computer <NUM>, query node computer <NUM>, and node computer <NUM>, distributed across one or more interconnected networks and may perform one or more of the method discussed herein.

A computer may be one or more software modules, computers, computer components, computing devices, databases, data stores, routers, switches, and/or other elements of internetworking infrastructure. Unless expressly stated as a "single" computer, a computer may be one or more computers.

More than one node may be executed on a single computer. For example, node computer <NUM> includes query node <NUM> and query node <NUM>, which are executed on node computer <NUM>, based on operating system <NUM>, and at least partially reside in memory device <NUM>.

While one or more of the components listed above may be illustrated as if running on a separate, remote computer from each other, one or more of the components listed above may be part of and/or executed on the same computer. For example, routing module <NUM>, record storage device <NUM>, source computer <NUM>, source node computer <NUM>, query node computer <NUM>, and/or any of the devices and/or modules included therein, may be executed on the same, single computer, local area network, and/or wide area network.

Routing computer <NUM> includes routing module <NUM>. Routing module <NUM> may receive one or more requests and/or queries from one or more client computers, generate one or more nodes, receive one or more records from one or more sources, nodes, modules, and/or computers, and/or send one or more records to the one or more other sources, nodes, modules, and/or computers. Additionally or alternatively, routing module <NUM> may store records in record storage device <NUM>. Additionally or alternatively, routing module <NUM> may send records and/or results to client computer <NUM> and/or any other modules, nodes, and/or computers.

A module may be software executed on a computer and/or a hardware that includes logic, and/or is coupled to, a computer. A module may include instructions and/or logic, which when executed cause performance of one or more of the methods discussed herein.

A routing module may receive and/or execute one or more continuous queries. A continuous query is a query for data, including a query against a database, which defines one or more functions to perform on streaming input data identified as one or more particular types. A continuous query may be defined by a query language, such as a structured query language or a continuous query language. A continuous query can be issued either as, or as part of, a database query or a data definition language command. The one or more functions defined by a continuous query may be database specific functions, networking functions, analytic functions, multimedia functions, and/or any other streaming data functions. For example, a continuous query may specify summarizing, aggregating, or filtering network analytics data in a network management application. A continuous query may define one or more data stream sources, schemas, types, subtypes, and/or super-types. A continuous query may define one or more operations assigned to one or more nodes. The continuous query may instruct the one or more nodes to wait for data to arrive, process the data, and then output derived, processed data. The features and processes discussed herein may be used for various purposes, such as streaming analytic data, debugging and diagnostics, recovery from failures, as well as in storing and processing previously received streaming data.

Routing module <NUM> may generate one or more types of nodes to execute one or more types of records based on one or more continuous queries. A first node may be a "first type of node" if the node processes a first type of record. Similarly, a second node may be a "second type of node" if the node processes a second type of record. In response to receiving a continuous query defining a first type of record received from source computer <NUM>, routing module <NUM> may generate, instantiate, and/or initialize source node computer <NUM>, query node computer <NUM>, and query node <NUM> on node computer <NUM>.

Routing module <NUM> may orchestrate which records are sent to which nodes and in which order. For example, routing module <NUM> may associate a first type of record, defined in a first continuous query, with the following nodes in the following order: source node computer <NUM>, query node computer <NUM>, and query node <NUM>. When a first type of record is received, routing module <NUM> may route the record to source node computer <NUM>. Based, at least in part, on routing module <NUM>, source node computer <NUM> may send the record, and/or data derived from the record, to query node computer <NUM>. Based, at least in part, on routing module <NUM>, query node computer <NUM> may send the record, and/or data derived from the record, to query node <NUM>. Based, at least in part on, routing module <NUM>, query node <NUM> may send the record, and/or data derived from the record, to client computer <NUM>, record storage device <NUM>, and/or any other computer and/or module defined by the continuous query. Accordingly, routing module <NUM> may comprise one or more tables, databases, and/or data structures for storing associations between one or more continuous queries, types of records, signatures, and/or nodes.

Routing module <NUM> may manage casting records from a subtype to a super-type and processing the "up-casted" record with the nodes associated with the super-type. For example, routing module <NUM> may receive a continuous query that defines a second type, and indicates that the second type is a subtype of a first type. In response, routing module <NUM> may generate, instantiate, and/or initialize the following nodes to process the second type of records in the following order: source node computer <NUM>, query node computer <NUM>, and query node <NUM>. In response to receiving a second type of record, routing module <NUM> may cast the second type of record as a first type of record, and/or send the up-casted record to the following nodes associated with the first type, in the following order: source node computer <NUM>, query node computer <NUM>, and query node <NUM>. Additionally or alternatively, based on routing module <NUM>, a node may up-cast a record and/or send the up-casted record to one or more nodes in a particular order.

Conventionally, when a record with an updated schema is received, each node previously processing the records may be suspended, reconfigured to process the new data, and resumed. As discussed herein, suspending one or more nodes can cause many problems. In contrast, routing module <NUM> is unconventional by instantiating new nodes to process the new data in a subtype or record, casting a subtype of record to a super-type of record, and routing the up-casted record to already-running nodes to process the data common between the subtype of record and the super-type of record, without suspending the already-running nodes. Thus, for these and many other reasons discussed herein, the unconventional methods and/or systems discussed herein, such as the methods and systems relating to routing module <NUM>, improve the functioning of the computer(s) and/or computer system(s).

A node may be a module, computer, computer component, computing device, database, data store, router, switch, and/or other element of internetworking infrastructure. A node may be defined by circuitry, one or more software instructions, a query, and/or a continuous query. A node may include logic and/or instructions, which when executed perform one or more of the methods discussed herein.

Each node may be configured to perform one or more different tasks, processes, and/or operations than another node. For example, a source node, such as source node computer <NUM>, may receive, format, enqueue, dequeue, and/or send records to one or more nodes, routing modules, and/or storage devices. Source node computer <NUM> may be specialized to process records that are a first type. Source node computer <NUM> may perform similar functions as source node computer <NUM>, but may be specialized to process records that are a second type.

Query node computer <NUM> may be configured to eliminate one or more first type records that do not satisfy one or more criteria, and send the records that do satisfy the one or more criteria to query node <NUM>. Query node <NUM> may perform one or more operations on the first type of records received from query node computer <NUM>, such as a sum, sort, and/or one or more other operations.

Query node computer <NUM> may perform one or more different operations on a second type of record, and/or data derived from a second type of record, than query node computer <NUM> processing an up-casted version of the same record, and/or data derived from the up-casted version of same record. Query node <NUM> may perform one or more different operations on a second type of record, and/or data derived from a second type of record, than query node <NUM> processing an up-casted version of the same record, and/or data derived from the up-casted version of same record.

A node that receives one or more records from one or more other nodes may be referred to as a "query node". For example, source node computer <NUM> may be a source node and query node computer <NUM> and query node <NUM> may each be a query node. However, in an embodiment, a source node may perform one or more of the operations that a query node may perform, and/or vice versa. For example, source node computer <NUM> may receive a record from source node computer <NUM>. Also for example, a source node computer <NUM> may receive and process a first type of record and need not send the record and/or data derived from the record to a downstream node.

A node may return results to routing module <NUM>, client computer <NUM>, and/or another module and/or computer. For example, query node <NUM> may receive one or more records indicating the CPU usage for one or more computers. Query node <NUM> may determine whether the CPU usage for each computer is greater than a particular threshold. If so, query node <NUM> may send a message to routing module <NUM> and/or client computer <NUM> indicating which computer(s) have a CPU usage greater than the particular threshold.

A node may add, remove, and/or change data in a record. For example, query node <NUM> may append and/or prepend data to each record indicating whether the record included data that satisfied one or more criteria. Query node <NUM> may send each modified record to routing module <NUM>, record storage device <NUM>, client computer <NUM>, and/or another node and/or computer. Also for example, source node computer <NUM> may receive a record, and/or a copy of a record, from source node computer <NUM>. Source node computer <NUM> may have already prepended a signature associated with the second type to the record. Source node computer <NUM> may determine that the signature is associated with the second type, determine that the second type is associated with the first type as a subtype based on routing module <NUM>, and/or replace the signature with a signature associated the first type.

A node may send a record, modified record, and/or data derived from a record to another node and/or computer. For example, source node computer <NUM> may send the record to query node computer <NUM> and/or routing module <NUM> for further processing. Also for example, routing module <NUM> may receive the record from source node computer <NUM>. Routing module <NUM> may modify the record and/or send the record to source node computer <NUM>.

In <FIG> a node may send and/or stream records to another node and/or computer directly. For example, source node computer <NUM> may be hardwired and/or hardcoded to send processed records to one or more nodes, such as query node computer <NUM>. Also for example, source node computer <NUM> may query routing module <NUM> for an address and/or identifier of the node that source node computer <NUM> should send a processed record to, based, at least in part, on a signature of the processed records and/or any other factor discussed herein. In this example, routing module <NUM> may comprise a relational database that associates the signature of the processed record with the address of query node computer <NUM>. Thus, routing module <NUM> may return the address of query node computer <NUM> to source node <NUM>, and source node <NUM> may send the process record query node computer <NUM>. Additionally or alternatively, a node may send and/or stream data to routing module <NUM> and/or record storage device <NUM>. Routing module <NUM> and/or record storage device <NUM> may push records to one or more nodes. Additionally or alternatively, a node may poll for data from routing module <NUM> and/or record storage device <NUM>. For example, query node computer <NUM> may query routing module <NUM> and/or record storage device <NUM> for records that source node computer <NUM> has processed and/or stored in record storage device <NUM>.

A source computer may be one or more computers streaming one or more particular types of records to one or more source nodes. For example, source computer <NUM> may be a first generation router on a first network, source computer <NUM> may be a first generation router on a second network, and source computer <NUM> may be a second generation router on a third network. Source computer <NUM> and source computer <NUM> may each send a plurality of records that are a first type of record to source node computer <NUM>. Source computer <NUM> may send a plurality of records that are a second type of record to source node computer <NUM>.

In <FIG> each of the source computers stream records directly to a source node. Additionally or alternatively, source computers may stream data to routing module <NUM> and/or record storage device <NUM>. Routing module <NUM> and/or record storage device <NUM> may push records to each source node. Additionally or alternatively, each source node may poll for data from routing module <NUM> and/or record storage device <NUM>.

Record storage device <NUM> may receive, store, and/or send one or more records. For example, source node computer <NUM> may receive a plurality of records from source computer <NUM>. Source node computer <NUM> may request and/or process the records. Source node computer <NUM> may send the record, and/or data derived from the record, to record storage device <NUM> to be stored. As query node computer <NUM> is ready to process the records, query node computer <NUM> may request and/or receive the records over time from record storage device <NUM>. Additionally or alternatively, routing module <NUM> may request one or more records in record storage device <NUM> and send the one or more records to a node and/or computer. Additionally or alternatively, routing module <NUM> may store one or more records and/or data sent to routing module <NUM> from one or more nodes and/or computers.

The systems, methods, and/or data structures discussed herein may be used to receive and process records with dynamic and/or evolving schemas without stopping and/or reconfiguring nodes that are already executing. <FIG> illustrates a process for processing a plurality of record types in one example embodiment. For purposes of illustrating a clear example, assume that a network administrator is using client computer <NUM> to analyze and monitor network connectivity between three networks. The first network is communicatively coupled with one or more networks through source computer <NUM>, which is a first generation router; the second network is communicatively coupled with one or more networks through source computer <NUM>, which is also first generation router; and the third network is communicatively coupled with one or more networks through source computer <NUM>, which is a second generation router. Records sent from a first generation router are a first type of record, records sent from a second generation router are a second type of record, and the second type is a subtype of the first type. Accordingly, each record that is a second type may include a "BGP Next Hop" field, whereas a first type of record need not.

In step <NUM>, a routing module associates a first signature with a first type and second signature with a second type. For example, the network administrator, through client computer <NUM>, may send two continuous queries to routing module <NUM>. The first continuous query may define a first signature for a first type of record to be the value signature <NUM>, and the second continuous query may define a second signature for a second type of record to be the value signature <NUM>. Accordingly, routing module <NUM> may update and/or store data in routing module <NUM>, routing computer <NUM>, a database, and/or any other computer coupled with routing module <NUM>, which associates signature <NUM> with the first type of record and signature <NUM> with the second type of record.

In step <NUM>, the routing module associates the first type with the first node and a second type with a second node. For example, the first continuous query may define one or more operations to be performed on each first type of record. Routing module <NUM> may analyze the one or more operations and generate, instantiate, and/or initialize source node computer <NUM> to receive one or more first type of records, query node computer <NUM> to perform a first operation defined in the first continuous query, and query node <NUM> to perform a sub-operation defined in the first continuous query. Similarly, the second continuous query may define one or more operations to be performed on each second type of record. Routing module <NUM> may analyze the one or more operations and generate, instantiate, and/or initialize source node computer <NUM> to receive one or more second type of records, query node computer <NUM> to perform a first operation defined in the second continuous query, and query node <NUM> to perform a sub-operation defined in the second continuous query.

In step <NUM>, the routing module associates the second type of record as a subtype of the first type of record. For example, the second continuous query may include data that expressly indicates that records of the second type are a subtype of the first type. In response, routing module <NUM> may associate the second signature (signature <NUM>) with the first signature (signature <NUM>), which indicates that records with the second signature are instances of a subtype of the first type. Additionally or alternatively, routing module <NUM> may analyze the field(s) expected to be read, and/or written to, by nodes initialized for the second continuous query, and may determine that those fields are a superset of the fields expected to be read, and/or written to, by nodes initialized for the first continuous query. Accordingly, routing module <NUM> may update and/or store data in routing module <NUM>, routing computer <NUM>, a database, and/or any other computer coupled with routing module <NUM>, which indicates that the second type is a subtype of the first type.

In step <NUM>, the first node and the second node receive a first record and a second record, respectively. For example, source computer <NUM> may send a first record, which may be a bare record and/or first type, to source node computer <NUM>. Source computer <NUM> may send a second record, which may be a bare record and/or second type, to source node computer <NUM>.

Additionally or alternatively, source computer <NUM> and source computer <NUM> may send the first record and the second record to routing module <NUM>. Routing module <NUM> may send the first record to source node computer <NUM> and the second record to source node computer <NUM>.

Additionally or alternatively, routing module <NUM> may store the records in record storage device <NUM>. Source node computer <NUM> may request records associated with the first type from routing module <NUM>. In response, routing module <NUM> may send the first record to source node computer <NUM>. Similarly, source node computer <NUM> may request records associated with the second type from routing module <NUM>. In response, routing module <NUM> may send the second record to source node computer <NUM>. Source node computer <NUM> and source node computer <NUM> may poll for records based on the one or more continuous queries, having finished processing one or more previously received records, an amount of time that has elapsed, available central processing unit cycles, available memory, and/or any other factor discussed herein and/or related to one or more computers executing source node computer <NUM>, source node computer <NUM>, routing module <NUM>, and/or record storage device <NUM>.

In step <NUM>, the first source node determines that the first record is a first type and the second source node determines that the second record is a second type. For example, source node computer <NUM> may determine that the first record is a first type based on the source, format, schema, and/or any other express and/or implied attribute(s) of the first record. Source node computer <NUM> may determine that the second record is a second type based on the source, format, schema, and/or any other express and/or implied attribute(s) of the second record. An express attribute may be any data and/or data structures included in the record, such as one or more field names. An implied attribute may be any data inferred from the record, such as the address of the source computer the record was received from, the make and/or model of the source computer the record was send from, and/or the time at which the record was received.

Additionally or alternatively, routing module <NUM> may determine that the first record is the first type and/or the second record is the second type using one or more of the factors discussed herein. In an embodiment, routing module <NUM> may determine the type of each record when the record is received.

In step <NUM>, the first source node associates the first signature with the first record and the second signature with the second record. For purposes of illustrating a clear example, assume that the first record is record <NUM> and the second record is record <NUM>; however, record <NUM> need not include signature <NUM> and record <NUM> need not include signature <NUM>. After receiving record <NUM>, source node computer <NUM> may prepend the first signature (signature <NUM>) to the record <NUM>, such that the signature <NUM> is the first datum in record <NUM>. After receiving record <NUM>, source node computer <NUM> may prepend the second signature (signature <NUM>) to record <NUM>, such that signature <NUM> is the first datum in record <NUM>.

Additionally or alternatively, routing module <NUM> may associate the first signature with the first record and the second signature with the second record by prepending the first signature to the first record and the second signature to the second record. Routing module <NUM> store each record in record storage device <NUM>. Each record stored by routing module <NUM> may include an express signature.

In step <NUM>, the first source node sends the first record to a first query node and the second source node sends the second record to a second query node. For example, source node computer <NUM> may send record <NUM> to query node computer <NUM>, and source node computer <NUM> may send record <NUM> to query node computer <NUM>. Query node computer <NUM> may process record <NUM> and query node computer <NUM> may process record <NUM>.

In step <NUM>, the source nodes may determine whether the record(s) they each received belongs to a subtype of another type. If so, then control may pass to step <NUM>. Otherwise, control may pass to step <NUM>. For example, source node computer <NUM> may determine that record <NUM> is not an instance of a subtype based on one or more of the following: signature <NUM> is not associated with another signature as a subtype; record <NUM> is not a superset of another type of record; and/or any other factor(s) discussed herein. Accordingly, source node computer <NUM> may proceed to step <NUM>.

Additional or alternatively, source node computer <NUM> may send record <NUM> to routing module <NUM>. Routing module <NUM> may determine whether record <NUM> is an instance of a subtype of another type of record based on one or more of the methods discussed herein. If so, then routing module <NUM> may pass to step <NUM>. Otherwise, routing module <NUM> may pass to step <NUM>.

As another example of step <NUM>, source node computer <NUM> may determine that record <NUM> is an instance of a super-type (the first type) based on one or more of the factors discussed herein, and proceed to step <NUM>. Additionally or alternatively, source node computer <NUM> may send record <NUM> to routing module <NUM>. Routing module <NUM> may determine record <NUM> is an instance of the first type based on one or more of the factors discussed herein, and proceed to step <NUM>.

In step <NUM>, the first source node and/or the routing module may associate the record with the super-type's signature. For example, source node computer <NUM> may receive a copy of the second record (record <NUM>) from source node computer <NUM> and/or routing module <NUM>. Record <NUM> may, but need not, be a bare record. Thus, record <NUM> may, but need not, include signature <NUM>. Source node computer <NUM> and/or routing module <NUM> may prepend signature <NUM>, and/or remove the signature <NUM>, from record <NUM> as illustrated in <FIG>.

In step <NUM>, the first source node may send the record with the super-type's signature to the node associated with the super-type's signature. For example, source node computer <NUM> and/or routing module <NUM> may send record <NUM> to the first query node associated with the first type: query node computer <NUM>. Control may then return to step <NUM>. If, for example, record <NUM> and/or record <NUM> is a subtype of a third type, then the process may be repeated. Otherwise, control may proceed to step <NUM>. In the current example, record <NUM> is not a subtype of an additional type, thus source node computer <NUM> and/or routing module <NUM> may proceed to step <NUM>.

In step <NUM>, control may terminate, wait for new records, and/or poll for new records. For example, source node computer <NUM> and/or routing module <NUM> may terminate, wait for one or more new records, and/or query and/or poll for one or more new records.

In the foregoing example, the records that were a second type and received from the second generation router (source computer <NUM>), were processed by both the plurality of nodes configured to process the second type of records (nodes <NUM>-<NUM>) and the plurality of nodes configured to process the first type of records (nodes <NUM>-<NUM>). Thus, source node computer <NUM>, query node computer <NUM>, and query node <NUM> could process the additional data in the second type of records (the "BGP Next Hop" field), and source node computer <NUM>, query node computer <NUM>, and query node <NUM> could still process each second type of record as if a first type of record without being suspended or reconfigured.

In the foregoing example, the first and the second continuous query were received before any record of any type was received. However, the second continuous query and/or the second type of records could be received after the first continuous query and/or the first type of records were received and/or began being processed by any node. Similarly, nodes <NUM>-<NUM> could have been generated, instantiated, and/or initialized after nodes <NUM>-<NUM> began processing the first type of records.

If a third generation router began streaming a third type of record to the system illustrated in <FIG>, and the third type of record was a subtype of the second type of record, then routing module <NUM> could generate, instantiate, and/or initialize one or more new nodes to process the additional data included in the third type of records. Routing module <NUM> may also cause each third type of record to be cast to a second type of record and processed by nodes <NUM>-<NUM>, using one or more of the methods discussed above, without suspending and/or reconfiguring nodes <NUM>-<NUM>. Additionally or alternatively, routing module <NUM> may also cause each third type of record to be cast to a first type of record and processed by nodes <NUM>-<NUM>, using one or more of the methods discussed above, without suspending and/or reconfiguring nodes <NUM>-<NUM>.

Using one or more of the methods discussed above, if a first type of record in a data stream was being processed according to a first continuous query by a first node, and the schema of the records in the data stream is updated to include a new field, then the routing module may send the records (which are now a second type, and a subtype of the first type) in the data stream to a second, new node. The second node may begin processing the data in the new field in each second type of record according to a second, new continuous query. The routing module may also up-cast each record as a first type of record and send each up-casted record to the first node, which is still executing the first continuous query. Thus, the first node may continue processing a first type of records and the new, second type of records without being suspended and/or reconfigured. Furthermore, each second type of record may be processed concurrently by the first node and the second node.

For example, <FIG> is a block diagram that illustrates a computer system <NUM> upon which an embodiment of the disclosure may be implemented.

Claim 1:
A method comprising:
receiving (<NUM>) a first record (<NUM>) comprising one or more first fields (114A) from a first source computer (<NUM>);
in response to determining (<NUM>) that the first record is a first type (<NUM>) and a first node (<NUM>) is associated (<NUM>) with the first type, sending (<NUM>) the first record to the first node, which is executed on a first computer (<NUM>), to be processed, wherein records of the first type are organized according to a first schema;
receiving (<NUM>) an updated first record (<NUM>) comprising the one or more first fields (114B) and one or more second, new fields (124A) from the first source computer;
in response to determining (<NUM>) that the updated first record is a second type (<NUM>) and a second node (<NUM>) is associated (<NUM>) with the second type, sending (<NUM>) the updated first record to the second node, which is executed on a second computer (<NUM>), to be processed, wherein records of the second type are organized according to a second schema;
in response to determining (<NUM>) that the second type is a first subtype of the first type, associating (<NUM>) the first updated record with a signature of the first type and sending (<NUM>) the first updated record to the first node to be processed, without suspending and/or reconfiguring the first node;
wherein the method is performed by one or more computing devices.