Skip to content

chore: Add tags to checkpoint #3291

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Fokko wants to merge 1 commit into
base: main
Choose a base branch
from
Open

chore: Add tags to checkpoint #3291

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 13 additions & 22 deletions kernel/src/checkpoint/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,17 +111,16 @@ use crate::action_reconciliation::{
ActionReconciliationIterator, ActionReconciliationIteratorState, RetentionCalculator,

@rliao147 rliao147 Sep 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);

};
use crate::actions::{
ADD_FIELD, CHECKPOINT_METADATA_NAME, DOMAIN_METADATA_FIELD, METADATA_FIELD, PROTOCOL_FIELD,
REMOVE_FIELD, SET_TRANSACTION_FIELD, SIDECAR_FIELD,
CheckpointMetadata, ADD_FIELD, CHECKPOINT_METADATA_FIELD, CHECKPOINT_METADATA_NAME,
DOMAIN_METADATA_FIELD, METADATA_FIELD, PROTOCOL_FIELD, REMOVE_FIELD, SET_TRANSACTION_FIELD,
SIDECAR_FIELD,
};
use crate::engine_data::FilteredEngineData;
use crate::expressions::{
lit, Expression, ExpressionRef, ExpressionStructPatchBuilder, Scalar, StructData,
};
use crate::expressions::{lit, Expression, ExpressionRef, ExpressionStructPatchBuilder, Scalar};
use crate::last_checkpoint_hint::LastCheckpointHint;
use crate::log_replay::LogReplayProcessor;
use crate::path::{self, ParsedLogPath};
use crate::schema::{lazy_schema_ref, schema, DataType, SchemaRef, StructField};
use crate::schema::{lazy_schema_ref, SchemaRef, StructField};
use crate::snapshot::SnapshotRef;
use crate::table_features::TableFeature;
use crate::table_properties::TableProperties;
Expand Down Expand Up @@ -327,20 +326,10 @@ fn base_checkpoint_action_fields() -> [&'static LazyLock<StructField>; 7] {
static CHECKPOINT_ACTIONS_SCHEMA_V1: LazyLock<SchemaRef> =
lazy_schema_ref! { ..(base_checkpoint_action_fields()) };

/// Schema for the checkpointMetadata field in V2 checkpoints.
/// We cannot use `CheckpointMetadata::to_schema()` as it would include the 'tags' field which
/// we're not supporting yet due to the lack of map support TODO(#880).
fn checkpoint_metadata_field() -> StructField {
StructField::nullable(
CHECKPOINT_METADATA_NAME,
schema! { not_null "version": LONG },
)
}

/// Schema for V2 checkpoints (includes checkpointMetadata action)
static CHECKPOINT_ACTIONS_SCHEMA_V2: LazyLock<SchemaRef> = lazy_schema_ref! {
..(base_checkpoint_action_fields()),
(checkpoint_metadata_field()),
(&CHECKPOINT_METADATA_FIELD),
};

/// Orchestrates the process of creating a checkpoint for a table.
Expand Down Expand Up @@ -692,11 +681,13 @@ impl CheckpointWriter {
// Start with an all-null row
let null_row = engine.evaluation_handler().null_row(schema.clone())?;

// Build the checkpointMetadata struct value
let checkpoint_metadata_value = Scalar::Struct(StructData::try_new(
vec![StructField::not_null("version", DataType::LONG)],
vec![Scalar::from(self.version)],
)?);
// Build the checkpointMetadata struct value. `tags` is left unset (null) since kernel
// does not currently emit checkpoint tags.
let checkpoint_metadata_value: Scalar = CheckpointMetadata {
version: self.version,
tags: None,
}
.into();

// Use a struct patch to set just the checkpointMetadata field, keeping others null
let patch = ExpressionStructPatchBuilder::new()
Expand Down
Loading