Consolidation Plan

Learn how to generate a consolidation plan and use it for consolidation.

How to run this tutorial

You can run this tutorial in two ways:

Locally on your machine.
On TileDB Cloud.

However, since TileDB Cloud has a free tier, we strongly recommend that you sign up and run everything there, as that requires no installations or deployment.

This tutorial demonstrates how to generate a consolidation plan and use it for consolidation. This tutorial uses a sparse array. Note that the functionality is not applicable to dense arrays. Before running this tutorial, it is recommended that you read the following sections:

First, import the necessary libraries, set the array URI (i.e., its path, which in this tutorial will be on local storage), and delete any previously created arrays with the same name.

Python

# Import necessary libraries
import tiledb
import numpy as np
import shutil
import os.path

# Set array URI
array_uri = os.path.expanduser("~/consolidation_plan")

# Delete array if it already exists
if os.path.exists(array_uri):
    shutil.rmtree(array_uri)

Next, create the array by specifying its schema.

Python

# Create the two dimensions
d1 = tiledb.Dim(name="d1", domain=(0, 3), tile=2, dtype=np.int32)
d2 = tiledb.Dim(name="d2", domain=(0, 3), tile=2, dtype=np.int32)

# Create a domain using the two dimensions
dom = tiledb.Domain(d1, d2)

# Create an attribute
a = tiledb.Attr(name="a", dtype=np.int32)

# Create the array schema with `sparse=True`.
sch = tiledb.ArraySchema(domain=dom, sparse=True, attrs=[a])

# Create the array on disk (it will initially be empty)
tiledb.Array.create(array_uri, sch)

Perform the first write.

Python

# Prepare some data in numpy arrays
d1_data = np.array([2, 0], dtype=np.int32)
d2_data = np.array([0, 1], dtype=np.int32)
a_data = np.array([4, 1], dtype=np.int32)

# Open the array in write mode and write the data in COO format
with tiledb.open(array_uri, "w") as A:
    A[d1_data, d2_data] = a_data

Perform the second write.

Python

# Prepare some data in numpy arrays
d1_data = np.array([1], dtype=np.int32)
d2_data = np.array([3], dtype=np.int32)
a_data = np.array([3], dtype=np.int32)

# Open the array in write mode and write the data in COO format
with tiledb.open(array_uri, "w") as A:
    A[d1_data, d2_data] = a_data

Perform the third write.

Python

# Prepare some data in numpy arrays
d1_data = np.array([2], dtype=np.int32)
d2_data = np.array([2], dtype=np.int32)
a_data = np.array([5], dtype=np.int32)

# Open the array in write mode and write the data in COO format
with tiledb.open(array_uri, "w") as A:
    A[d1_data, d2_data] = a_data

Inspect the array folder to confirm that three fragments have been created.

Python

/Users/stavrospapadopoulos/consolidation_plan
├── __commits
│   ├── __1718414474463_1718414474463_0c1f5a89c19deaa17071686bcc64d4a8_21.wrt
│   ├── __1718414474471_1718414474471_72a3e0642cac75221a6e4fa0dcf6a825_21.wrt
│   └── __1718414474478_1718414474478_236806f61ff15d9ffecca1b38d920513_21.wrt
├── __fragment_meta
├── __fragments
│   ├── __1718414474463_1718414474463_0c1f5a89c19deaa17071686bcc64d4a8_21
│   │   ├── __fragment_metadata.tdb
│   │   ├── a0.tdb
│   │   ├── d0.tdb
│   │   └── d1.tdb
│   ├── __1718414474471_1718414474471_72a3e0642cac75221a6e4fa0dcf6a825_21
│   │   ├── __fragment_metadata.tdb
│   │   ├── a0.tdb
│   │   ├── d0.tdb
│   │   └── d1.tdb
│   └── __1718414474478_1718414474478_236806f61ff15d9ffecca1b38d920513_21
│       ├── __fragment_metadata.tdb
│       ├── a0.tdb
│       ├── d0.tdb
│       └── d1.tdb
├── __labels
├── __meta
└── __schema
    ├── __1718414474458_1718414474458_00000002acef5200321fb6f42db433cf
    └── __enumerations

11 directories, 16 files

Read the array data. Observe that the data from both writes are present in the result, as expected.

Python

# Read array
with tiledb.open(array_uri, "r") as A:
    print(A[:])

OrderedDict([('a', array([1, 3, 4, 5], dtype=int32)), ('d1', array([0, 1, 2, 2], dtype=int32)), ('d2', array([1, 3, 0, 2], dtype=int32))])

Generate the consolidation plan. That creates a single set of fragments (i.e, a single “node”, representing the set of fragments that should be submitted to a single worker for consolidation).

Python

# Get consolidation plan
A = tiledb.open(array_uri, "r")
fragment_size = 10000  # desirable fragment size in bytes
cons_plan = tiledb.ConsolidationPlan(tiledb.default_ctx(), A, fragment_size)

# Output information about the consolidation plan
print(cons_plan.dump())
print(cons_plan)
print("\n")
print("Number of nodes: ", cons_plan.num_nodes)
print("Number of fragments in node #1: ", cons_plan[0]["num_fragments"])
print("Fragments URIs in node #1", cons_plan[0]["fragment_uris"])

# Close the array
A.close()

{
  "nodes": [
    {
      "uris" : [
        {
           "uri" : "__1718414474463_1718414474463_0c1f5a89c19deaa17071686bcc64d4a8_21"
        },
        {
           "uri" : "__1718414474478_1718414474478_236806f61ff15d9ffecca1b38d920513_21"
        }
      ]
    }
  ]
}

{'fragments': {'node_0': {'fragment_uris': ['__1718414474463_1718414474463_0c1f5a89c19deaa17071686bcc64d4a8_21',
                                            '__1718414474478_1718414474478_236806f61ff15d9ffecca1b38d920513_21'],
                          'num_fragments': 2}},
 'num_nodes': 1}


Number of nodes:  1
Number of fragments in node #1:  2
Fragments URIs in node #1 ['__1718414474463_1718414474463_0c1f5a89c19deaa17071686bcc64d4a8_21', '__1718414474478_1718414474478_236806f61ff15d9ffecca1b38d920513_21']

Now run consolidation, passing the fragment URIs from the consolidation plan. Also perform vacuuming, so that the two original fragments that participated in consolidation get deleted.

Python

# Consolidate using the fragments from the consolidation plan
tiledb.consolidate(
    array_uri,
    config=tiledb.Config({"sm.consolidation.mode": "fragments"}),
    fragment_uris=cons_plan[0]["fragment_uris"],
)

# Vacuum
tiledb.vacuum(array_uri, config=tiledb.Config({"sm.vacuum.mode": "fragments"}))

Inspecting the file hierarchy of the array again, observe that only two fragments exist: the fragment from the first write (which is intact), and the fragment generated by consolidating the fragments determined by the consolidation plan.

Python

/Users/stavrospapadopoulos/consolidation_plan
├── __commits
│   ├── __1718414474463_1718414474478_6144a5572364b9b6c03799a8ae3d1642_21.wrt
│   └── __1718414474471_1718414474471_72a3e0642cac75221a6e4fa0dcf6a825_21.wrt
├── __fragment_meta
├── __fragments
│   ├── __1718414474463_1718414474478_6144a5572364b9b6c03799a8ae3d1642_21
│   │   ├── __fragment_metadata.tdb
│   │   ├── a0.tdb
│   │   ├── d0.tdb
│   │   ├── d1.tdb
│   │   └── t.tdb
│   └── __1718414474471_1718414474471_72a3e0642cac75221a6e4fa0dcf6a825_21
│       ├── __fragment_metadata.tdb
│       ├── a0.tdb
│       ├── d0.tdb
│       └── d1.tdb
├── __labels
├── __meta
└── __schema
    ├── __1718414474458_1718414474458_00000002acef5200321fb6f42db433cf
    └── __enumerations

10 directories, 12 files

Read the array again and observe that nothing changed. The same results are present there.

Python

# Read array
with tiledb.open(array_uri, "r") as A:
    print(A[:])

OrderedDict([('a', array([1, 3, 4, 5], dtype=int32)), ('d1', array([0, 1, 2, 2], dtype=int32)), ('d2', array([1, 3, 0, 2], dtype=int32))])

Clean up in the end by deleting the array.

Python

# Delete the array
if os.path.exists(array_uri):
    shutil.rmtree(array_uri)