System and method for storing stream data in distributed relational tables with data provenance

• US 9,390,147 B2
• Filed: 09/24/2012
• Issued: 07/12/2016
• Est. Priority Date: 09/23/2011
• Status: Active Grant

• Current Assignee: Mr. Yusefzadeh, Bahram Nmn
• Original Assignee: Red Lambda Incorporated
• Inventors: Bird, Robert, Leko, Adam, Whitlock, Matthew
• Primary Examiner(s): Cao, Phuong Thao
• Application Number:
  US13/625,434
• Publication Number:
  US 20130080393A1
• Time in Patent Office: 1,387 Days
• Field of Search: 707/649, 707/748
• US Class Current: 1/1
• CPC Class Codes: G06F 16/219 Managing data history or versioning querying versioned data G06F16/2474; querying temporal data G06F16/2477
• CPC Class Codes: G06F 16/2477 Temporal data queries
• CPC Class Codes: G06F 2201/84 Using snapshots, i.e. a logical point-in-time copy of the data

Abstract

A system, a method and a computer readable medium for storing data elements and related data provenance information. The data elements may be represented in a hyper-table having rows and columns which may be indexed. The data-values of the corresponding data-elements in the hyper-cells may be retrieved based on the indices. Snapshots of the indices may be generated at pre-determined time periods. Checkpoints of the hyper-table may be generated at time periods that are based on transactions on the hyper-table. The hyper-table is capable of being queried as the hyper-table existed at certain time-periods, and data-values of the data-elements may be retrieved as the data-elements existed at such time-periods.

Claims

1. A method for storing and querying data, comprising the steps of:

representing data-elements in a hyper-table stored in a table-store, wherein the data-elements are allocated to data-blocks stored in the table-store, wherein the table-store is located on a distributed device, wherein the hyper-table comprises;

(i) hyper-rows representing the data-elements allocated to the data-blocks, (ii) at least one hyper-column associated with an attribute of the corresponding data-elements, and (iii) hyper-cells having data-values of the corresponding data-elements;

generating indices of the hyper-rows and the at least one hyper-column for the corresponding data-elements, wherein the indices are located on the distributed device and the data-values of the corresponding data-elements in the hyper-cells are capable of being retrieved based on the indices;

storing a pre-determined time period on the distributed device;

generating snapshots of the indices at each occurrence of the pre-determined time-period, wherein the snapshots comprise the indices as the indices existed at the occurrence of the pre-determined time-period;

generating checkpoints of the hyper-table at each occurrence of the pre-determined time-period; and

selecting one of a plurality of occurrences of the pre-determined time period and querying the snapshots of the indices and the checkpoints of the hyper-table to return query results based on the state of the indices and the hyper-table as of the selected occurrence of the pre-determined time-period,

wherein said distributed device comprises a processor, a random-access memory, and a network interface connected to a network and is connected to a plurality of remote distributed devices, each of which comprises a processor, a random-access memory and a network interface connected to the network and is specially configured to store at least a portion of the data elements and at least a portion of the indices.

2. The method of claim 1, wherein the pre-determined time period is based on fixed time intervals.

3. The method of claim 1, wherein the pre-determined time period is based on data-values of a plurality of the data-elements in a transaction.

4. The method of claim 1, wherein each attribute of the corresponding data-elements comprises a data-name and a data-type for the corresponding data-elements.

5. The method of claim 1, further comprising the step of:

assigning block-identifiers to the data-blocks.

6. The method of claim 5, wherein the data-blocks comprise binary large objects, wherein each binary large object is associated with an unique address, wherein the unique address is based on content of the binary large object, wherein the assigned block-identifiers are based on the corresponding unique addresses.

7. The method of claim 1, further comprising the steps of:

performing a transaction on the hyper-table;

generating a binary large object corresponding to the transaction;

generating a checkpoint of the hyper-table, wherein the checkpoint tracks a delta, wherein the delta corresponds to the binary large object; and

recording the time of the transaction.

8. The method of claim 7, wherein the transaction is selected from a group consisting of a hyper-row insertion, a hyper-row deletion, and a hyper-row modification.

9. The method of claim 7, wherein the hyper-table comprises a plurality of distributed hyper-tables, wherein the transaction is performed on only one of the plurality of distributed hyper-tables, wherein referential integrity of the only one of the plurality of distributed hyper-tables is maintained.

10. The method of claim 7, wherein the transaction corresponds to a hyper-row in the hyper-table, wherein the generated binary large object corresponding to the transaction comprises a data-value of a data-element represented by the hyper-row, wherein the data-value comprises changed-values and unchanged-values, and wherein the changed-values are based on the transaction.

11. The method of claim 1, further comprising the step of:

sequentially numbering each checkpoint for each one of a plurality of distributed hyper-tables, wherein the checkpoints are ordered for each one of the plurality of distributed hyper-tables.

12. The method of claim 11, wherein the sequential numbering of each checkpoint is generated so that conflicts are prevented.

13. The method of claim 1, further comprising the steps of:

generating recovered-indices based on most-recent snapshots, wherein the most-recent snapshots comprise the indices that were generated at a recent time-period, wherein the recovered-indices are generated for recovery purposes after a system-failure;

loading the recovered-indices; and

applying post-snapshot checkpoints to the recovered-indices, wherein the post-snapshot checkpoints are the checkpoints generated at the time-periods subsequent to the recent time-period, wherein the recovered-indices are modified based on the post-snapshot checkpoints.

14. The method of claim 1, further comprising the steps of:

executing a query for the data-elements represented in the hyper-table based on at least one of the checkpoints;

generating a result set of hyper-rows based on the query; and

applying subsequent-checkpoints to the result set of hyper-rows, wherein the subsequent-checkpoints are the checkpoints generated at the time-periods subsequent to the time-period of the at least one of the checkpoints, wherein the result set of hyper-rows are modified based on the subsequent-checkpoints.

15. The method of claim 1, further comprising the steps of:

executing a query for the data-elements represented in the hyper-table based on at least one of the snapshots of the indices;

generating a result set of hyper-rows based on the query; and

applying subsequent-checkpoints to the result set of hyper-rows, wherein the subsequent-checkpoints are the checkpoints generated at the time-periods subsequent to the pre-determined time-period of the at least one of the snapshots, wherein the result set of hyper-rows are modified based on the subsequent-checkpoints.

16. The method of claim 1, further comprising the steps of:

executing a query for the data-elements represented in the hyper-table based on at least one of the snapshots of the indices, wherein the pre-determined time-period of the at least one of the snapshots is subsequent to a time-period of at least one of the checkpoints;

generating a result set of hyper-rows based on the query; and

applying prior-checkpoints to the result set of hyper-rows, wherein the prior-checkpoints are the checkpoints generated at the time-periods prior to the pre-determined time-period of the at least one of the snapshots and subsequent to the time-period of the at least one of the checkpoints, wherein the result set of hyper-rows are modified based on the prior-checkpoints, wherein the modified result set of hyper-rows represents the data-elements as the associated the data-elements existed at the time-period of the at least one of the checkpoints.

17. The method of claim 1, further comprising the steps of:

logically joining at least two of a plurality of hyper-tables based on predetermined rules, wherein the predetermined rules comprise join-operations;

generating join-views based on the joined at least two of the plurality of hyper-tables; and

indexing the join-views, wherein the join-views are hyper-tables, wherein each of the join-views has hyper-rows representing the data-elements.

18. The method of claim 17, further comprising the steps of:

performing a join-operation on two or more hyper-tables, wherein the two or more hyper-tables are selected from a group consisting of the plurality of hyper-tables and the join-views, wherein the join-views are hyper-tables based on at least two of the plurality of hyper-tables that are logically joined; and

storing a result set of hyper-rows based on the performed join-operation, wherein the stored result set is stored as a checkpoint in a joining hyper-table.

19. The method of claim 18, further comprising the step of:

generating a checkpoint in the joining hyper-table only when identifying a delta in the joining hyper-table identified based on a transaction on the hyper-table.

20. The method of claim 1, further comprising the steps of:

retrieving a plurality of consecutive-checkpoints, wherein the consecutive-checkpoints are a consecutive subset of the checkpoints, wherein each checkpoint tracks a delta; and

cumulatively applying the tracked deltas of the plurality of consecutive-checkpoints to a result set of hyper-rows, the result set of hyper-rows generated from the query for the data-elements represented in the hyper-table based on at least one of the snapshots of the indices.

21. The method of claim 1, wherein the hyper-table maintain a set of listeners, wherein the listeners are notified upon generation of a checkpoint for the hyper-table.

22. The method of claim 1, wherein the table-store comprises a plurality of distributed table-stores located on a plurality of distributed devices, wherein the plurality of distributed devices are connected via a network, wherein at least one of the data-blocks is stored on a first distributed device, and wherein at least one of the indices is stored on a second distributed device.

23. The method of claim 22, wherein the first distributed device is the second distributed device.

24. The method of claim 22, wherein the first distributed device and the second distributed device are different devices.

25. The method of claim 1, further comprising the steps of:

determining a current data-value for at least one data-element;

determining a historic data-value for the at least one data-element;

comparing the current data-value and the historic data-value; and

generating comparisons based on the current data-value and the historic data-value, wherein the comparisons indicate value-changes of the at least one data-element for a time-period.

26. The method of claim 1, wherein the hyper-table has a schema, wherein the schema comprises a data-name and a data-type for the data-elements represented in the hyper-table.

27. The method of claim 1, wherein the table-store is a distributed file system.

28. The method of claim 1, wherein the hyper-table comprises a physical data structure and a logical structure, wherein the indices are stored in the logical structure.

29. The method of claim 1, wherein at least one of the indices comprises a metric tree logical structure.

30. The method of claim 1, wherein at least one of the data-elements is received via data-streams from a plurality of distributed devices connected by a network.

31. A system for storing data, comprising:

a distributed device, a table-store on said distributed device adapted to store data-blocks, the table-store having data-elements allocated to the data-blocks, the data-elements represented in a hyper-table, the hyper-table comprises;

(i) hyper-rows representing the data-elements allocated to the data-blocks, (ii) at least one hyper-column associated with an attribute of the corresponding data-elements, and (iii) hyper-cells having data-values of the corresponding data-elements;

indices generated by said system of the hyper-rows and the at least one hyper-column for the corresponding data-elements;

a pre-determined time period stored on the distributed device;

snapshots generated by said system of the indices at each occurrence of the pre-determined time-period, the snapshots comprising the indices as the indices existed at the occurrence of the pre-determined time-period, the snapshots generated at each occurrence of the pre-determined time-period;

checkpoints generated by said system of the hyper-table as stored on the table-store at each occurrence of the pre-determined time-period; and

selections of one of a plurality of occurrences of the pre-determined time period,

whereby a query of the snapshots of the indices and the checkpoints of the hyper-table may return query results based on the state of the indices and the hyper-table as of the selected occurrence of the pre-determined time-period,

wherein said distributed device comprises a processor, a random-access memory, and a network interface connected to a network and is connected to a plurality of remote distributed devices, each of which comprises a processor, a random-access memory and a network interface connected to the network and is specially configured to store at least a portion of the data elements and at least a portion of the indices.

32. A non-transitory computer readable medium having computer readable instructions stored thereon for execution by a processor, wherein the instructions on the non-transitory computer readable medium are adapted to:

represent data-elements in a hyper-table stored in a table-store, wherein the data-elements are allocated to data-blocks stored in the table-store, wherein the table-store is located on a distributed device, wherein the hyper-table comprises;

(i) hyper-rows representing the data-elements allocated to the data-blocks, (ii) at least one hyper-column associated with an attribute of the corresponding data-elements, and (iii) hyper-cells having data-values of the corresponding data-elements;

store a pre-determined time period on the distributed device;

generate indices of the hyper-rows and the at least one hyper-column for the corresponding data-elements, wherein the indices are located on the distributed device;

generate snapshots of the indices at each occurrence of the pre-determined time-period, wherein the snapshots comprise the indices as the indices existed at the occurrence of the pre-determined time-period;

generate checkpoints of the hyper-table at each occurrence of the pre-determined time-period; and

select one of a plurality of occurrences of the pre-determined time period and query the snapshots of the indices and the checkpoints of the hyper-table to return query results based on the state of the indices and the hyper-table as of the selected occurrence of the pre-determined time-period,

wherein said distributed device comprises a processor, a random-access memory, and a network interface connected to a network and is connected to a plurality of remote distributed devices, each of which comprises a processor, a random-access memory and a network interface connected to the network and is specially configured to store at least a portion of the data elements and at least a portion of indices.