feat: add Snapshot::checkpoint() & Table::checkpoint() API

[sebastiantia]

What changes are proposed in this pull request?

Motivation and Context

This is part of the parent-issue to provide checkpoint write support in delta-kernel-rs #499. This PR introduces a new snapshot API for writing single-file checkpoints to Delta tables (#736), supporting single-file V1 and V2 spec classic-named checkpoint formats.

Breaking Changes

• Introduces a new error variant that may require handling in downstream code.

Summary of Changes

• New Checkpoint API: Adds Snapshot::checkpoint & Table::checkpoint as the entry points for checkpoint creation, supporting both V1 and V2 checkpoint specs depending on table features.
• CheckpointWriter: Introduces a core orchestrator for checkpoint creation, including data preparation and (future) finalization - todo(feat: add CheckpointWriter::finalize() #850)
• CheckpointDataIterator: An iterator-based mechanism for checkpoint data generation, ensuring accurate action statistics and safe finalization.
• Explicit Finalization: Lays groundwork for a two-step checkpoint process. First write all data, then call .finalize() to persist checkpoint metadata - todo(feat: add CheckpointWriter::finalize() #850).
• Error Handling: Adds a new error variant for checkpoint write failures.
• Extensibility: The API is designed to accommodate future enhancements, ( multi-file V2 checkpoints )

Major Components and APIs

API / Struct	Description
Error::CheckpointWrite(String)	New error variant for checkpoint write failures
Snapshot::checkpoint()	Creates a new CheckpointWriter for a snapshot
Table::checkpoint()	Creates a new CheckpointWriter for a version of a table
CheckpointWriter	Main class orchestrating checkpoint creation and finalization
CheckpointWriter::checkpoint_path()	Returns the URL where the checkpoint file should be written
CheckpointWriter::checkpoint_data()	Returns the checkpoint data (CheckpointDataIterator) to be written to the checkpoint file
CheckpointWriter::finalize()	todo(#850) Finalizes checkpoint by writing the _last_checkpoint file after data is persisted
CheckpointDataIterator	Iterator over checkpoint actions, accumulates statistics for finalization
CheckpointBatch (private)	Output of CheckpointLogReplayProcessor, contains filtered actions and counts

Checkpoint Types

Table Feature	Resulting Checkpoint Type
No v2Checkpoints	Single-file Classic-named V1
v2Checkpoints	Single-file Classic-named V2

• V1: For legacy tables, no CheckpointMetadata action included.
• V2: For tables supporting v2Checkpoints, includes CheckpointMetadata action for enhanced metadata.

Usage Workflow

1. Create a CheckpointWriter via Snapshot::checkpoint or Table::checkpoint
2. Retrieve the checkpoint path from CheckpointWriter::checkpoint_path()
3. Retrieve the checkpoint data from CheckpointWriter::checkpoint_data()
4. Write the data to path in object storage (engine-specific)
5. Finalize the checkpoint by calling CheckpointWriter.finalize() - todo(feat: add CheckpointWriter::finalize() #850)

todo(#850): The CheckpointWriter::finalize() API that was previously included in this PR has been split into a separate PR #851 for ease of review. Handle the finalization of the checkpointing process by writing the _last_checkpoint file on call to.finalize(). Note: we require the engine to write the entire checkpoint file to storage before calling .finalize(), otherwise the table may be corrupted. It will be hard for the engine not to do this since the finalize() call takes the FileMeta of the checkpoint write

How was this change tested?

Unit tests in checkpoint/mod.rs
test_deleted_file_retention_timestamp - tests file retention timestamp calculations
test_create_checkpoint_metadata_batch

Unit tests in checkpoint/tests.rs
test_v1_checkpoint_latest_version_by_default: table that does not support v2Checkpoint, no checkpoint version specified
test_v1_checkpoint_specific_version: table that does not support v2Checkpoint, checkpointing at a specific version
test_v2_checkpoint_supported_table: table that supports v2Checkpoint & no version is specified

[scovich]

Getting very very close, looking very nice

[scovich]

If we wanted to, we could still expose the old Table::checkpoint method, which takes an impl Into<Option<Version>>. Rationale: I would have directly checkpointed my snapshot if I had one handy, but I don't, so I have to call into the Table instead. So it's actually helpful for the method on Table to combine the snapshot+checkpoint operations.

These unit tests do a pretty good job of illustrating the scenario.

[sebastiantia]

👍 I'll expose the old Table::checkpoint method as well

[scovich]

aside: This seems to come up more and more lately -- we really need a utility method that returns "now" in Delta timestamp format, that everyone can use instead of having to replicate the multi-step conversion process.

[sebastiantia]

Tracked 👍 #877

[scovich]

So this test is pretending that "now" is 1000 seconds [after what]?

tiny nit:

Suggested change

	let result =
	deleted_file_retention_timestamp_with_time(retention, Duration::from_secs(1000))?;
	let now = Duration::from_secs(1000);
	let result = deleted_file_retention_timestamp_with_time(retention, now)?;

[sebastiantia]

Good question — in this test, now is a simulated timestamp representing 1000 seconds since the UNIX epoch It doesn’t matter in practice since the test just fixes now at 1000s past UNIX epoch to keep the math predictable. The logic under test only cares about relative time (i.e., now minus retention)

[sebastiantia]

After a second look, I cleaned up the test case to be more readable

[scovich]

Very nice!

[zachschuermann]

LGTM pending last few questions/comments!

[zachschuermann]

Also looks like this is only used below - should we just make self.create_checkpoint_metadata_batch only take engine? and internally just use self.version?