Keep writer state across batches

[robinskil]

Fixes #106

This pull request introduces a mechanism for tracking and updating key entry states within the AsyncOdvWriter in beacon-arrow-odv/src/writer.rs. The main goal is to ensure that key columns in profile and time series batches are updated with unique suffixes based on their entry key and state, improving batch classification and data integrity. The changes include new state-tracking structs, integration of these states into batch classification logic, and updates to how key columns are suffixed.

Key Entry State Management

• Added FileWriterState and KeyEntryState structs to track the current key and its state for profile and time series batches. This includes logic for equality and ordering of key states to determine when to increment suffixes.
• Integrated optional trajectory_profile_key_entry_state and trajectory_time_series_key_entry_state fields into the AsyncOdvWriter struct to cache and manage key entry states across batch writes.

Batch Classification Logic

• Modified the classify_batch and related methods to accept and update the new key entry state caches, ensuring that key suffixes are generated based on changes in entry key and state. [1] [2] [3]
• Updated the logic for appending suffixes to key columns: now the suffix reflects the count from the key entry state, incrementing only when the entry key or state changes.

API and Usability Improvements

• Changed method signatures and internal variable names for clarity and consistency, such as renaming time_column to time_column_name. [1] [2] [3]

These changes collectively improve the accuracy and uniqueness of key columns in profile and time series batches, making the writer more robust for downstream data consumers.

[Copilot]

Pull Request Overview

This PR implements stateful tracking for key entries in the ODV writer to ensure unique key column suffixes across batches. The main purpose is to maintain writer state between batch writes, preventing duplicate key suffixes when processing profile and time series data.

Key Changes:

• Added FileWriterState and KeyEntryState structs to track entry keys and their states across batch writes
• Modified batch classification logic to use cached states and generate unique key suffixes based on state changes
• Renamed time_column parameter to time_column_name for consistency

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

The use core::time; import appears to be unused. Consider removing this import unless it's required elsewhere in the file.

Suggested change

use core::time;

[Copilot]

Comparing a boolean value with == false is unnecessary. Use the ! operator instead: !eq_array.as_any().downcast_ref::<arrow::array::BooleanArray>().unwrap().value(0)

Suggested change

	if eq_array
	.as_any()
	.downcast_ref::<arrow::array::BooleanArray>()
	.unwrap()
	.value(0)
	== false
	if !eq_array
	.as_any()
	.downcast_ref::<arrow::array::BooleanArray>()
	.unwrap()
	.value(0)

[Copilot]

The pattern of checking is_none() followed by a match on as_mut() creates redundant logic. Since you initialize the state if it's None on line 779, the None => 0 case on line 802 is unreachable. Consider removing the None arm or restructuring to use get_or_insert_with.

Suggested change

	let count = match profile_file_state.as_mut() {
	Some(state) => {
	if state.is_equal_entry_key(&entry_key_scalar) {
	if state.is_equal_key_state(&current_entry_key_state) {
	state.key_state.count
	} else {
	state.key_state.count += 1;
	state.key_state.count
	}
	} else {
	state.entry_key = entry_key_scalar;
	state.key_state =
	KeyEntryState::new(vec![timestamp_scalar.clone()]);
	state.key_state.count
	}
	}
	None => 0,
	let state = profile_file_state.as_mut().unwrap();
	let count = if state.is_equal_entry_key(&entry_key_scalar) {
	if state.is_equal_key_state(&current_entry_key_state) {
	state.key_state.count
	} else {
	state.key_state.count += 1;
	state.key_state.count
	}
	} else {
	state.entry_key = entry_key_scalar;
	state.key_state =
	KeyEntryState::new(vec![timestamp_scalar.clone()]);
	state.key_state.count