Pqax.Collector 0.2.2

Pqax.Collector

Collects the footprint of Delta tables, and of folders of Parquet files with no table format, and writes it as a PQAX file. It reads the Delta log of each table and, at collection level 2, the footer of each Parquet file. It never reads row data, and it keeps no data value unless asked. The format is defined by the PQAX serialization and the PQAX object model; the file is written with the Pqax library.

The package is not published yet. Reference the project src/dotnet/Pqax.Collector/Pqax.Collector.csproj.

Collect

using Pqax;
using Pqax.Collector;
using Pqax.Collector.ParquetNet;

var storage = new LocalStorage(@"conformance\fixtures\lakehouse");   // the folder plays the workspace
var tables = new List<TableLocation>();
foreach (string item in await TableDiscovery.ListItemsAsync(storage, CancellationToken.None))
{
    tables.AddRange(await TableDiscovery.DiscoverAsync(storage, item, CancellationToken.None));
}

var collector = new PqaxCollector(storage, new CollectorOptions(), new ParquetNetCheckpointReader());
FootprintBuilder footprint = await collector.CollectAsync(tables, progress: null, CancellationToken.None);
footprint.Collection.CollectorApp = "MyTool";
footprint.Write("lakehouse.pqax");

For OneLake, the storage takes a delegate that returns an access token for the scope https://storage.azure.com/.default, so that the library does not depend on Azure.Identity. The application signs in with any TokenCredential of Azure.Identity, or gets the token another way:

// Azure.Identity, in the application: any TokenCredential.
TokenCredential credential = new ClientSecretCredential(tenantId, clientId, Environment.GetEnvironmentVariable("PQAX_CLIENT_SECRET"));
// or new ClientCertificateCredential(tenantId, clientId, "collector.pfx"), new ManagedIdentityCredential(ManagedIdentityId.SystemAssigned),
// new AzureCliCredential(), new DefaultAzureCredential(), new InteractiveBrowserCredential()

using var storage = new OneLakeStorage("My Workspace", async cancellation =>
    (await credential.GetTokenAsync(new TokenRequestContext(new[] { OneLakeStorage.Scope }), cancellation)).Token);
IReadOnlyList<TableLocation> tables = await TableDiscovery.DiscoverAsync(storage, "Sales.Lakehouse", CancellationToken.None);

The delegate is called before every request. The credentials of Azure.Identity cache their tokens, but a delegate that calls another source should cache the token itself until shortly before it expires, as TokenCache of Pqax.Cli does. A token acquired elsewhere, such as the one of a notebook, is passed as it is: new OneLakeStorage("My Workspace", _ => Task.FromResult(token)).

A service principal or a managed identity needs the Fabric tenant setting that lets service principals use Fabric APIs, and a workspace role that reads OneLake: Contributor or higher works, and so does Viewer with the items shared with the principal; Viewer alone does not. The Pqax.Cli README has the details. A refused request is a StorageException with StatusCode 401 or 403 and IsAccessDenied true.

On a real workspace of 78 tables and 5,236 files, a collection at level 2 with every footer took 7,623 requests, one per footer, and wrote a 1.8 MB file. With a fixed 32 KB tail it read 177 MB in 2 minutes; with the adaptive tail it read 103 MB in 38 seconds, with the same output.

The workspace and the items are named both by name or both by id. OneLake answers 400 FriendlyNameSupportDisabled to a workspace id followed by item names.

A container of a storage account with hierarchical namespace, Azure Data Lake Storage Gen2, is read by the same class with the endpoint of the account: Azure Storage serves the DFS API that OneLake serves, with the container where OneLake has the workspace. The token is for the same scope, and the identity needs the role Storage Blob Data Reader on the container or the account. DiscoverFolderAsync finds the tables under a folder of the container, as Folders of a storage account says:

using var storage = new OneLakeStorage("lake", tokenProvider,
    new OneLakeStorageOptions { Endpoint = new Uri("https://contoso.dfs.core.windows.net") });   // the container "lake"
IReadOnlyList<TableLocation> tables = await TableDiscovery.DiscoverFolderAsync(storage, "sales/curated", CancellationToken.None);

pqax collect in Pqax.Cli does both.

Public API

Options

OneLakeStorageOptions sets how the storage connects:

What it reads

For each table:

For a folder of Parquet files, the format Parquet, there is no log: one listing of the folder gives the files, the footer of the most recent file gives the columns at every level, and at level 2 every other footer is read too, the first one not again. See Folders of Parquet files.

Every request goes through the IStorage, so the collection states its requests and bytes. When the storage implements IStorageStatistics, those numbers come from it and include retries and the pages of a listing.

The log is replayed by the rules of the Delta protocol: a file is identified by its path and its deletion vector, and the latest action wins. A table whose checkpoint is not read keeps what the JSON commits say: when the log still has every commit from version 0, the collector replays them and the table is complete; otherwise its files are unknown, its file totals are null, it gets the issue DeltaCheckpointNotRead, and it has the Summary depth.

Commits

The collector chooses the commit files to read from the latest version back:

1. Every commit that the state needs: those after the checkpoint, or all of them when the state comes from the JSON commits alone. The limit does not apply to them.
2. Then older commits, the most recent first, until the commit files read reach MaxCommitsPerTable or the log has no older one.

So a table reads the greater of the limit and the commits that the state needs, within the commits that the log keeps. A limit of 0 reads only the commits that the state needs. commitsRead is the number of commit files read. The commits of the table, kept from the Files depth, hold every one of them, and the operations and oldestCommitAt cover them all. A commit file whose read fails counts toward the limit, since it cost a request, and gets an issue instead of a place in commitsRead and in the commits.

What the footprint holds

The collector fills every field of the object model that a Delta table and its footers give, and follows the rule that unknown is not zero: a value that was not read is null, a total with an unknown term is null, and a known empty list is empty.

Partitions and commits are kept from the Files depth, as in the prototype.

Unreferenced files

Locations

TableDiscovery fills the location of each table, and the location and the identifiers of its source, from the storage and from the item as the caller names it.

• The workspace and the item are written as the caller gave them, without escaping. <host> is the host of the endpoint, onelake.dfs.fabric.microsoft.com by default.
• An identifier is recorded only for a workspace or an item given as an id: a GUID with hyphens. A listing of the DFS endpoint has no item id, so with names the identifiers stay null: the collector does not guess them.
• The tables with the same source kind, name, and location share one source. A TableLocation built by hand records what its constructor is given.
• A local location is a full path, so it can hold the name of the user of the machine. The object model marks the locations and the identifiers as names, for the obfuscation to replace.

Folders of a storage account

TableDiscovery.DiscoverFolderAsync(storage, folder, maxDepth) walks a folder of a storage that has no item layout, such as a container of Azure Data Lake Storage Gen2, with one listing for each folder walked:

Every table it finds has one source:

Folders of Parquet files

A table of the format Parquet has no log, so the footprint holds what the files and their footers say:

Issues

Footer read failures

A footer that cannot be read becomes an issue with the path of its file, and the collection goes on. The code depends on the failure:

A table with files and no footer read has the collection level 1 and at most the Files depth.

Footer parser

The parser reads the Thrift compact protocol: field headers with short and long deltas, booleans in the header, zigzag varints, doubles, binaries, lists, sets, maps, structures, and UUIDs. It skips a field that it does not know, of any type, and a known field with an unexpected type. It decodes FileMetaData, SchemaElement with its logical and converted types, RowGroup, ColumnChunk, ColumnMetaData, Statistics, PageEncodingStats, SizeStatistics, the bloom filter, column index, and offset index lengths, and the key-value metadata.

It refuses an encrypted footer (PARE), a file that does not end with PAR1, a footer length that does not fit the file, a footer over 64 MB, a required field that is missing, a row group whose column chunks do not match the leaves of the schema, and any length or count that goes beyond the bytes left. It never throws anything but ParquetFooterException on bad bytes.

Limits

• Collection levels 3 and 4 and Iceberg tables are not supported in this version.
• A folder of Parquet files is read only when its files are directly in it: a partitioned folder, with <column>=<value> subfolders, gives no table. Its partition values would come from the paths, which this version does not read.
• DiscoverFolderAsync stops at 4 levels below its folder by default. A container with many folders above its tables costs one listing for each.
• A storage account is read through its DFS endpoint, which needs hierarchical namespace. A flat Blob Storage account, Amazon S3, and Databricks volumes are not supported in this version.
• A checkpoint larger than 512 MB is not read.
• The log compaction files, <x>.<y>.compacted.json, are counted with the log and not used.
• All the files of a table are kept in memory until the footprint is built, as row objects in the builder.