chore: Add tags to checkpoint

[Fokko]

What changes are proposed in this pull request?

By adding tags to checkpoint, we can leverage CheckpointMetadata::to_schema(), simplifying the code.

How was this change tested?

Existing tests

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 90.28%. Comparing base (d634b50) to head (6bc125f).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #3291   +/-   ##
=======================================
  Coverage   90.28%   90.28%           
=======================================
  Files         251      251           
  Lines       88575    88572    -3     
  Branches    88575    88572    -3     
=======================================
- Hits        79973    79971    -2     
+ Misses       5718     5717    -1     
  Partials     2884     2884           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

Benchmark results: ✅ Pass

Summary: 🚀 0  ·  ✅ 13  ·  ☑️ 2  ·  🚧 0  ·  ❌ 0

Per-benchmark results (15 rows)

Test	Change	Base	PR
clustered/readMetadataLatestPredicate/serial	✅ 1.04x faster	70.3±1.78ms	67.8±2.70ms
crcLatest/snapshotLatest	✅ 1.10x faster	8.6±1.13ms	7.8±0.85ms
crcMissing/snapshotLatest	✅ 1.00x	20.7±1.77ms	20.8±1.71ms
crcSlightlyStale/snapshotLatest	☑️ 1.02x slower	9.1±1.05ms	9.3±0.95ms
crcVeryStale/snapshotLatest	✅ 1.01x faster	13.8±1.32ms	13.6±1.35ms
partitioned/readMetadataLatestPredicate/serial	✅ 1.03x faster	43.3±2.50ms	42.2±1.93ms
v1Checkpoint/readMetadataLatest/serial	☑️ 1.03x slower	9.7±0.52ms	10.0±0.54ms
v1Checkpoint/snapshotLatest	✅ 1.01x faster	640.4±59.22µs	633.0±58.83µs
v2Checkpoint/readMetadataLatest/parallel2	✅ 1.01x faster	6.9±0.24ms	6.8±0.26ms
v2Checkpoint/readMetadataLatest/serial	✅ 1.02x faster	11.4±0.22ms	11.2±0.21ms
v2Checkpoint/snapshotLatest	✅ 1.02x faster	607.2±57.75µs	597.8±59.84µs
wideSchemaJsonStats/readMetadataLatestPredicate/serial	✅ 1.11x faster	68.8±3.85ms	62.1±2.50ms
wideSchemaJsonStats/snapshotLatest	✅ 1.10x faster	1830.1±108.92µs	1669.5±49.83µs
wideSchemaStructStats/readMetadataLatestPredicate/serial	✅ 1.09x faster	30.0±1.08ms	27.4±0.90ms
wideSchemaStructStats/snapshotLatest	✅ 1.04x faster	1667.3±89.65µs	1608.1±38.40µs

_{Legend: 🚀 ≥1.15x faster  · ✅ faster or unchanged  · ☑️ ≤1.03x slower  · 🚧 1.03x-1.15x slower  · ❌ ≥1.15x slower
Commit: 6bc125f · Trigger: auto-push · Tags: base · Updated: 2026-09-09 08:28 PDT}

[rliao147]

Thanks for the contribution, I can stamp after the test changes are done. Also, please check merge conflicts.

[rliao147]

can you please add these validations to test_create_checkpoint_metadata_batch ?

Now, we're writing tags to disk when Kernel writes a V2 checkpoint file, but we're lacking some regression tests to validate the tags are correct.

  let cm = record_batch.column_by_name("checkpointMetadata").unwrap()
      .as_any().downcast_ref::<StructArray>().unwrap();
  let tags = cm.column_by_name("tags").expect("checkpointMetadata must carry a tags field");
  assert!(tags.as_any().downcast_ref::<MapArray>().is_some(), "tags must be a map");
  assert!(tags.is_null(0), "tags should be written null");
  let version = cm.column_by_name("version").unwrap()
      .as_any().downcast_ref::<Int64Array>().unwrap();
  assert_eq!(version.value(0), snapshot.version() as i64);