re_log_encoding 0.38.1

use std::collections::{BTreeMap, HashMap};

use arrow::array::RecordBatch;
use arrow::buffer::NullBuffer;
use arrow::datatypes::Field;
use arrow::{
    array::{BooleanArray, UInt64Array},
    error::ArrowError,
};
use itertools::Itertools as _;
use re_chunk::external::nohash_hasher::IntMap;
use re_chunk::external::re_byte_size;
use re_chunk::{ArchetypeName, ChunkError, ChunkId, ComponentIdentifier, ComponentType, Timeline};
use re_log_types::{AbsoluteTimeRange, EntityPath, StoreId, TimeType, TimelineName};
use re_types_core::{
    ComponentDescriptor, FIELD_METADATA_KEY_ARCHETYPE, FIELD_METADATA_KEY_COMPONENT,
    FIELD_METADATA_KEY_COMPONENT_TYPE,
};

use crate::{CodecError, CodecResult, Decodable as _, StreamFooterEntry, ToApplication as _};

/// The payload found in [`super::RrdFooter`]s.
///
/// Each `RrdManifest` corresponds to one, and exactly one, RRD stream (i.e. recording).
/// This restriction exists to make working with multiple RRD streams much simpler: due to the way
/// the Rerun data model works, filtering rows of data from a manifest can have hard-to-predict
/// second order effects on the schema of the stream as a whole.
/// By keeping manifests for different recordings separate, we remove the need to filter per
/// recording ID, greatly simplifying the process.
///
/// This is an application-level type, the associated transport-level type can be found
/// over at [`re_protos::log_msg::v1alpha1::RrdManifest`].
///
/// ## What's in the box?
///
/// Each row in this manifest describes a unique chunk (ID, offset, size, timeline & component stats, etc).
/// This can be used to compute relevancy queries (latest-at, range, dataframe), without needing to load
/// any of the actual data in memory.
///
/// You can think of an RRD Manifest as a dataframe of the time panel, effectively.
///
/// The best way to understand what an RRD manifest does or doesn't do is to look at snapshots for
/// a simple recording:
/// ```text,ignore
/// ┌───────────────────────────────────────────────┬──────────────────────────────────┬──────────────────────────────────┐
/// │ chunk_entity_path                             ┆ /my_entity1                      ┆ /my_entity1                      │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ chunk_id                                      ┆ 00000000000000010000000000000001 ┆ 00000000000000010000000000000002 │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ chunk_is_static                               ┆ false                            ┆ true                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ chunk_num_rows                                ┆ 4                                ┆ 1                                │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ chunk_byte_offset                             ┆ 0                                ┆ 962                              │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ chunk_byte_size                               ┆ 962                              ┆ 464                              │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ chunk_byte_size_uncompressed                  ┆ 3981                             ┆ 2509                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ example_MyPoints:colors:has_static_data       ┆ false                            ┆ true                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ example_MyPoints:labels:has_static_data       ┆ false                            ┆ true                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ example_MyPoints:points:has_static_data       ┆ false                            ┆ false                            │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:start                            ┆ PT10S                            ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:end                              ┆ PT40S                            ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:start                                ┆ 10                               ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:end                                  ┆ 40                               ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:start                                ┆ 1970-01-01T00:00:00.000000010    ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:end                                  ┆ 1970-01-01T00:00:00.000000040    ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:example_MyPoints:colors:start    ┆ PT20S                            ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:example_MyPoints:colors:end      ┆ PT30S                            ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:example_MyPoints:colors:num_rows ┆ 2                                ┆ 0                                │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:example_MyPoints:points:start    ┆ PT10S                            ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:example_MyPoints:points:end      ┆ PT40S                            ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ elapsed_time:example_MyPoints:points:num_rows ┆ 3                                ┆ 0                                │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:example_MyPoints:colors:start        ┆ 20                               ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:example_MyPoints:colors:end          ┆ 30                               ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:example_MyPoints:colors:num_rows     ┆ 2                                ┆ 0                                │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:example_MyPoints:points:start        ┆ 10                               ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:example_MyPoints:points:end          ┆ 40                               ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ frame_nr:example_MyPoints:points:num_rows     ┆ 3                                ┆ 0                                │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:example_MyPoints:colors:start        ┆ 1970-01-01T00:00:00.000000020    ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:example_MyPoints:colors:end          ┆ 1970-01-01T00:00:00.000000030    ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:example_MyPoints:colors:num_rows     ┆ 2                                ┆ 0                                │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:example_MyPoints:points:start        ┆ 1970-01-01T00:00:00.000000010    ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:example_MyPoints:points:end          ┆ 1970-01-01T00:00:00.000000040    ┆ null                             │
/// ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
/// │ log_time:example_MyPoints:points:num_rows     ┆ 3                                ┆ 0                                │
/// └───────────────────────────────────────────────┴──────────────────────────────────┴──────────────────────────────────┘
/// ```
///
/// Note that `:start` & `:end` columns are always implicitly _inclusive_. The `_inclusive` suffix has been
/// removed to reduce noise.
///
/// ## Understand size/offset columns
///
/// * `chunk_byte_size` & `chunk_byte_offset` are always reported using the backend's native
///   storage size. For a backend that makes use of compression, such as an RRD file with
///   compression enabled, these sizes are therefore compressed. For a backend that doesn't do any
///   kind of compression, such as the OSS server that stores everything already decoded in memory,
///   these sizes will correspond to heap memory usage.
/// * `chunk_byte_size_uncompressed` reports the chunk's uncompressed size, whose exact meaning
///   depends on the writer: the length of the uncompressed Arrow IPC stream when written to an RRD
///   file, and the decoded heap size when synthesized in memory
///   (see [`RawRrdManifest::build_in_memory_from_chunks`]).
/// * `chunk_key`, if specified, should always be used to fetch the associated data.
///
/// ## A note on filtering
///
/// Always be on your toes when filtering rows out of an RRD manifest. Due to how the Rerun data
/// model works, removing rows (and therefore chunks) from a recording can also affect the number
/// of columns in that recording (because e.g. all the data for a specific entity path is gone),
/// which in turn will affect the Sorbet schema of the recording too.
///
/// Filtering RRD manifests is very non trivial and should only be performed with great care.
#[derive(Clone, Debug, re_byte_size::SizeBytes)]
pub struct RawRrdManifest {
    /// The recording ID that was used to identify the original recording.
    ///
    /// This is extracted from the `SetStoreInfo` message of the associated RRD stream.
    pub store_id: StoreId,

    /// The Sorbet schema of the recording, following the usual merging and sorting rules.
    ///
    /// ⚠️ This is the Sorbet schema of the recording being indexed by this manifest, *not* the
    /// schema of [`Self::data`].
    pub sorbet_schema: arrow::datatypes::Schema,

    /// The SHA256 hash of the Sorbet schema of the associated RRD stream.
    ///
    /// This is always computed by sorting the fields of the schema by name first.
    /// See [`RrdManifest::compute_sorbet_schema_sha256]`.
    pub sorbet_schema_sha256: [u8; 32],

    /// The actual manifest data, which catalogs every chunk in this recording.
    ///
    /// Each row in this dataframe describes a unique chunk (ID, offset, size, timeline & component stats, etc).
    /// This can be used to compute relevancy queries (latest-at, range, dataframe), without needing to load
    /// any of the actual data in memory.
    ///
    /// Note in particular that there isn't any recording ID column in here, since an [`RawRrdManifest`]
    /// is always scoped to a single recording: the one specified in [`RawRrdManifest::store_id`].
    //
    // TODO(cmc): should we slap a sorbet:version on this? that probably should be part of the sorbet ABI as
    // much as anything else?
    pub data: arrow::array::RecordBatch,
}

/// A map based representation of the static data within an [`RawRrdManifest`].
pub type RrdManifestStaticMap = IntMap<EntityPath, IntMap<ComponentIdentifier, ChunkId>>;

/// The individual entries in an [`RrdManifestTemporalMap`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, re_byte_size::SizeBytes)]
pub struct RrdManifestTemporalMapEntry {
    /// The time range covered by this entry.
    pub time_range: AbsoluteTimeRange,

    /// The number of rows in the original chunk which are associated with this entry.
    ///
    /// At most, this is the same as the number of rows in the chunk as a whole. For a specific
    /// entry it might be less, since chunks allow sparse components.
    pub num_rows: u64,
}

/// A map based representation of the temporal data within an [`RawRrdManifest`].
pub type RrdManifestTemporalMap = IntMap<
    EntityPath,
    IntMap<Timeline, IntMap<ComponentIdentifier, BTreeMap<ChunkId, RrdManifestTemporalMapEntry>>>,
>;

/// The sha256 hash of an [RRD manifest's payload].
///
/// Can be used as a unique and/or content-addressable ID.
///
/// Refer to [`RawRrdManifest::compute_sha256`] to see how exactly this is computed.
///
/// [RRD manifest's payload]: `RawRrdManifest::data`
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RrdManifestSha256(pub [u8; 32]);

impl std::fmt::Display for RrdManifestSha256 {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "RrdManifest#{}", sha256_to_hex(&self.0))
    }
}

/// Format a SHA256 hash as a lowercase hex string.
pub fn sha256_to_hex(sha256: &[u8; 32]) -> String {
    use std::fmt::Write as _;

    sha256
        .iter()
        .fold(String::with_capacity(2 * sha256.len()), |mut hex, byte| {
            write!(hex, "{byte:02x}").ok();
            hex
        })
}

impl RawRrdManifest {
    /// Concatenate multiple manifests by appending all rows together.
    ///
    /// All manifests must be for the same recording (same `store_id` and sorbet schema).
    /// The data `RecordBatch`es are concatenated.
    ///
    /// This is used when the server sends a manifest in multiple parts.
    pub fn concat(manifests: &[&Self]) -> Result<Self, ArrowError> {
        re_tracing::profile_function!();

        let first = manifests.first().ok_or_else(|| {
            ArrowError::InvalidArgumentError("No manifests to concatenate".to_owned())
        })?;

        for other in &manifests[1..] {
            if first.store_id != other.store_id {
                return Err(ArrowError::SchemaError(
                    "Mismatching store_id in RawRrdManifest::concat".to_owned(),
                ));
            }

            if first.sorbet_schema_sha256 != other.sorbet_schema_sha256 {
                return Err(ArrowError::SchemaError(
                    "Mismatching sorbet recording schemas in RawRrdManifest::concat".to_owned(),
                ));
            }

            if first.data.schema() != other.data.schema() {
                re_log::debug!(
                    "Different schemas in the RrdManifest ({} columns in existing, {} in the new part)",
                    first.data.num_columns(),
                    other.data.num_columns(),
                );
            }
        }

        let batches: Vec<&RecordBatch> = manifests.iter().map(|m| &m.data).collect();
        let data = arrow::compute::concat_batches(&first.data.schema(), batches)?;

        Ok(Self {
            store_id: first.store_id.clone(),
            sorbet_schema: first.sorbet_schema.clone(),
            sorbet_schema_sha256: first.sorbet_schema_sha256,
            data,
        })
    }

    /// Returns this manifest without the chunks logged under `__properties`.
    ///
    /// An asset joins the store of the segment that references it, where its own recording
    /// properties would take the place of that segment's.
    pub fn without_recording_properties(self) -> CodecResult<Self> {
        re_tracing::profile_function!();

        let keep: BooleanArray = self
            .col_chunk_entity_path()?
            .iter()
            .map(|entity_path| Some(!EntityPath::parse_forgiving(entity_path).is_property()))
            .collect();

        if keep.true_count() == keep.len() {
            return Ok(self);
        }

        let data = arrow::compute::filter_record_batch(&self.data, &keep)
            .map_err(CodecError::ArrowDeserialization)?;

        Ok(Self { data, ..self })
    }

    /// Merges multiple manifests into one, tolerating schema differences.
    ///
    /// Unlike [`Self::concat`] — which requires identical schemas and `store_id`s — this allows
    /// each input to carry its own schema. Fields that exist in some manifests but not others are
    /// padded with nulls during merging. Sorbet schemas are unified via
    /// [`arrow::datatypes::Schema::try_merge`] and the hash is recomputed.
    ///
    /// The resulting manifest's `store_id` is set to the provided value — the individual inputs'
    /// `store_id`s are ignored, since this method is designed for cases where the inputs come from
    /// distinct recordings (e.g. per-layer manifests being merged into a segment-scoped one).
    ///
    /// The `:has_static_data` and `:num_rows` columns are non-nullable in the `RawRrdManifest`
    /// contract: cross-layer null values are collapsed to the underlying buffer default
    /// (`false` / `0`) so that clients get the same shape regardless of whether a column came
    /// from one layer or many. This mirrors what `ExtendedRrdManifest::try_into_standard_manifest`
    /// does on the commercial side.
    pub fn merge(store_id: StoreId, manifests: Vec<Self>) -> CodecResult<Self> {
        re_tracing::profile_function!();

        if manifests.is_empty() {
            return Err(CodecError::ArrowDeserialization(
                ArrowError::InvalidArgumentError("cannot merge 0 manifests".to_owned()),
            ));
        }

        let parts: Vec<_> = manifests
            .into_iter()
            .map(|m| (m.sorbet_schema, m.data))
            .collect();

        let (sorbet_schema, sorbet_schema_sha256, data) = Self::merge_polymorphic_parts(parts)?;

        let data = strip_null_mask_on_default_columns(data)?;

        Ok(Self {
            store_id,
            sorbet_schema,
            sorbet_schema_sha256,
            data,
        })
    }

    /// Polymorphic-merge kernel: unify sorbet schemas and concatenate data batches.
    ///
    /// Takes one `(sorbet_schema, data_batch)` pair per logical input manifest, and returns
    /// `(merged_sorbet_schema, merged_sha256, merged_data_batch)`.
    pub fn merge_polymorphic_parts(
        parts: Vec<(arrow::datatypes::Schema, RecordBatch)>,
    ) -> CodecResult<(arrow::datatypes::Schema, [u8; 32], RecordBatch)> {
        use re_arrow_util::{RecordBatchExt as _, concat_polymorphic_batches};

        let (sorbet_schemas, data_batches): (Vec<_>, Vec<_>) = parts.into_iter().unzip();

        let sorbet_schema = arrow::datatypes::Schema::try_merge(sorbet_schemas)
            .map_err(CodecError::ArrowDeserialization)?;
        let sorbet_schema_sha256 = Self::compute_sorbet_schema_sha256(&sorbet_schema)
            .map_err(CodecError::ArrowSerialization)?;

        // Make all fields nullable so `concat_polymorphic_batches` can backfill missing columns
        // with nulls.
        let nullable_batches: Vec<RecordBatch> = data_batches
            .into_iter()
            .map(|b| b.make_nullable())
            .collect();

        let data = concat_polymorphic_batches(&nullable_batches)
            .map_err(CodecError::ArrowDeserialization)?;

        Ok((sorbet_schema, sorbet_schema_sha256, data))
    }

    /// High-level helper to parse [`RawRrdManifest`]s from raw RRD bytes.
    ///
    /// This does not decode all the data, but rather goes straight to the RRD footer (if any).
    ///
    /// * Returns `None` if no valid footer was found.
    /// * Returns an error if either the footer or any of the manifests are corrupt.
    ///
    /// Usage:
    /// ```text,ignore
    /// let rrd_bytes = std::fs::read("/path/to/my/recording.rrd");
    /// let rrd_manifests = RrdManifest::from_rrd_bytes(&rrd_bytes)?;
    /// let rrd_manifest = rrd_manifests
    ///     .into_iter()
    ///     .find(|m| m.store_id.kind() == StoreKind::Recording)?;
    /// ```
    pub fn from_rrd_bytes(rrd_bytes: &[u8]) -> CodecResult<Vec<Self>> {
        let stream_footer = match crate::StreamFooter::from_rrd_bytes(rrd_bytes) {
            Ok(footer) => footer,

            // That was in fact _not_ a footer.
            Err(CodecError::FrameDecoding(_)) => return Ok(vec![]),

            Err(err) => Err(err)?,
        };

        let mut manifests = Vec::new();

        for entry in stream_footer.entries {
            let StreamFooterEntry {
                rrd_footer_byte_span_from_start_excluding_header,
                crc_excluding_header,
            } = entry;

            let rrd_footer_byte_span = rrd_footer_byte_span_from_start_excluding_header;

            let rrd_footer_byte_span = rrd_footer_byte_span
                .try_cast::<usize>()
                .ok_or_else(|| {
                    CodecError::FrameDecoding(
                        "RRD footer too large for native bit width".to_owned(),
                    )
                })?
                .range();

            let rrd_footer_bytes = &rrd_bytes[rrd_footer_byte_span];

            let crc = crate::StreamFooter::compute_crc(rrd_footer_bytes);
            if crc != crc_excluding_header {
                return Err(CodecError::CrcMismatch {
                    expected: crc_excluding_header,
                    got: crc,
                });
            }

            let rrd_footer =
                re_protos::log_msg::v1alpha1::RrdFooter::from_rrd_bytes(rrd_footer_bytes)?;
            let new_manifests: Vec<_> = rrd_footer
                .manifests
                .iter()
                .map(|manifest| manifest.to_application(()))
                .try_collect()?;
            manifests.extend(new_manifests);
        }

        Ok(manifests)
    }

    /// This builds an [`RawRrdManifest`] from a collection of chunks, assuming an in-memory backend.
    ///
    /// This is only useful for developement/testing purposes.
    /// All sanity checking methods, including the costly ones, will be called before returning.
    ///
    /// Chunk offsets will start at 0 and increment from there according to their heap size.
    /// There are no chunk keys whatsoever.
    pub fn build_in_memory_from_chunks<'a>(
        store_id: StoreId,
        chunks: impl Iterator<Item = &'a re_chunk::Chunk>,
    ) -> CodecResult<Self> {
        let mut rrd_manifest_builder = crate::RrdManifestBuilder::default();

        let mut offset = 0;
        for chunk in chunks {
            let chunk_batch = chunk.to_chunk_batch()?;

            use re_byte_size::SizeBytes as _;
            let byte_size_uncompressed = chunk.heap_size_bytes();

            let uncompressed_byte_span =
                re_span::Span::from_start_len(offset, byte_size_uncompressed);

            offset += byte_size_uncompressed;

            rrd_manifest_builder.append(
                &chunk_batch,
                uncompressed_byte_span,
                byte_size_uncompressed,
            )?;
        }

        let rrd_manifest = rrd_manifest_builder.build(store_id)?;
        rrd_manifest.sanity_check_cheap()?;
        rrd_manifest.sanity_check_heavy()?;

        Ok(rrd_manifest)
    }

    /// Computes the sha256 hash of the manifest's data, which can be used as a unique ID.
    //
    // TODO(cmc): very expensive, should be cached somewhere.
    pub fn compute_sha256(&self) -> Result<RrdManifestSha256, ArrowError> {
        re_tracing::profile_function!();

        let data_ipc = {
            let mut data_ipc = Vec::new();
            let mut w =
                arrow::ipc::writer::StreamWriter::try_new(&mut data_ipc, self.data.schema_ref())?;
            w.write(&self.data)?;
            data_ipc
        };

        use sha2::Digest as _;
        let mut hash = [0u8; 32];
        let mut hasher = sha2::Sha256::new();
        hasher.update(&data_ipc);
        hasher.finalize_into(sha2::digest::generic_array::GenericArray::from_mut_slice(
            &mut hash,
        ));

        Ok(RrdManifestSha256(hash))
    }

    /// Computes a map-based representation of the static data in this RRD manifest.
    pub fn calc_static_map(&self) -> CodecResult<RrdManifestStaticMap> {
        re_tracing::profile_function!();

        use re_arrow_util::ArrowArrayDowncastRef as _;

        let mut per_entity: RrdManifestStaticMap = IntMap::default();

        let chunk_ids = self.col_chunk_id_iter()?;
        let chunk_entity_paths = self.col_chunk_entity_path_iter()?;
        let chunk_is_static = self.col_chunk_is_static_iter()?;

        let has_static_component_data: Vec<_> =
            itertools::izip!(self.data.schema_ref().fields(), self.data.columns(),)
                .filter(|(f, _c)| Self::is_index_has_static_data(f))
                .map(|(f, c)| {
                    c.try_downcast_array_ref::<arrow::array::BooleanArray>()
                        .map_err(|err| {
                            CodecError::ArrowDeserialization(ArrowError::SchemaError(format!(
                                "cannot downcast column '{}': {err}",
                                f.name(),
                            )))
                        })
                        .map(|c| (f, c))
                })
                .try_collect()?;

        for (i, (chunk_id, is_static, entity_path)) in
            itertools::izip!(chunk_ids, chunk_is_static, chunk_entity_paths).enumerate()
        {
            if !is_static {
                continue;
            }

            for (f, has_static_component_data) in &has_static_component_data {
                let has_static_component_data = has_static_component_data.value(i);
                if !has_static_component_data {
                    continue;
                }

                let Some(component) = f.metadata().get(FIELD_METADATA_KEY_COMPONENT) else {
                    return Err(CodecError::from(ChunkError::Malformed {
                        reason: format!(
                            "column '{}' is missing rerun:component metadata",
                            f.name()
                        ),
                    }));
                };
                let component = ComponentIdentifier::try_new(component).map_err(|err| {
                    CodecError::from(ChunkError::Malformed {
                        reason: err.to_string(),
                    })
                })?;

                let per_component = per_entity.entry(entity_path.clone()).or_default();

                // TODO(cmc): technically we should follow the usual crazy semantics to decide which
                // static chunk for which component in case of conflicts etc but, it's fine for now.
                per_component
                    .entry(component)
                    .and_modify(|id| *id = chunk_id)
                    .or_insert(chunk_id);
            }
        }

        Ok(per_entity)
    }

    /// Computes a map-based representation of the temporal data in this RRD manifest.
    pub fn calc_temporal_map(&self) -> CodecResult<RrdManifestTemporalMap> {
        re_tracing::profile_function!();

        use re_arrow_util::ArrowArrayDowncastRef as _;

        let fields = self.data.schema_ref().fields();
        let columns = self.data.columns();
        let indexes = fields
            .iter()
            .filter_map(|f| {
                f.metadata()
                    .get(Self::FIELD_METADATA_KEY_INDEX)
                    .and_then(|index| {
                        f.metadata()
                            .get(FIELD_METADATA_KEY_COMPONENT)
                            .map(|c| (index, c, f))
                    })
            })
            .filter(|(_index, _component, field)| Self::is_index_start(field))
            .collect_vec();

        let mut per_entity: RrdManifestTemporalMap = Default::default();

        let chunk_ids = self.col_chunk_id_iter()?;
        let chunk_entity_paths = self.col_chunk_entity_path_iter()?;
        let chunk_is_static = self.col_chunk_is_static_iter()?;

        struct IndexColumns<'a> {
            index: &'a str,
            component: &'a String,
            time_type: TimeType,

            col_start_nulls: NullBuffer,
            col_start_raw: &'a [i64],

            col_end_nulls: NullBuffer,
            col_end_raw: &'a [i64],

            col_num_rows_raw: &'a [u64],
        }

        // Two descriptors that share a component identifier but differ in type or archetype get
        // two sets of columns with the same names, so a column is matched to its `:start` field
        // by its full metadata, never by name or by `rerun:component` alone.
        let sibling = |field: &Field, marker: &str| {
            itertools::izip!(fields, columns)
                .find(|(f, _col)| {
                    Self::has_index_marker(f, marker) && f.metadata() == field.metadata()
                })
                .ok_or_else(|| {
                    CodecError::from(ChunkError::Malformed {
                        reason: format!("{marker} index is missing for {}", field.name()),
                    })
                })
        };

        let mut columns_per_index = Vec::<IndexColumns<'_>>::new();
        for (index, component, field) in indexes {
            let index = index.as_str();
            if index == Self::INDEX_NAME_STATIC {
                continue;
            }

            let (_, col_start) = sibling(field, Self::INDEX_MARKER_START)?;
            let (_, col_end) = sibling(field, Self::INDEX_MARKER_END)?;
            let (field_num_rows, col_num_rows) = sibling(field, Self::INDEX_MARKER_NUM_ROWS)?;

            let (time_type, col_start_raw) =
                TimeType::from_arrow_array(col_start).map_err(CodecError::ArrowDeserialization)?;
            let (_, col_end_raw) =
                TimeType::from_arrow_array(col_end).map_err(CodecError::ArrowDeserialization)?;
            let col_num_rows_raw: &[u64] = col_num_rows
                .try_downcast_array_ref::<UInt64Array>()
                .map_err(|err| {
                    CodecError::ArrowDeserialization(ArrowError::SchemaError(format!(
                        "cannot downcast column '{}': {err}",
                        field_num_rows.name(),
                    )))
                })?
                .values();

            // So we don't have to pay the virtual call cost for every `is_valid()` call.
            let col_start_nulls = col_start
                .nulls()
                .cloned()
                .unwrap_or_else(|| NullBuffer::new_valid(col_start.len()));
            let col_end_nulls = col_end
                .nulls()
                .cloned()
                .unwrap_or_else(|| NullBuffer::new_valid(col_end.len()));

            columns_per_index.push(IndexColumns {
                index,
                component,
                time_type,
                col_start_nulls,
                col_start_raw,
                col_end_nulls,
                col_end_raw,
                col_num_rows_raw,
            });
        }

        for (i, (chunk_id, is_static, entity_path)) in
            itertools::izip!(chunk_ids, chunk_is_static, chunk_entity_paths).enumerate()
        {
            if is_static {
                continue;
            }

            for columns in &columns_per_index {
                let IndexColumns {
                    index,
                    component,
                    time_type,
                    col_start_nulls,
                    col_start_raw,
                    col_end_nulls,
                    col_end_raw,
                    col_num_rows_raw,
                } = columns;

                if !col_start_nulls.is_valid(i) || !col_end_nulls.is_valid(i) {
                    continue;
                }

                let component = ComponentIdentifier::try_new(component).map_err(|err| {
                    CodecError::from(ChunkError::Malformed {
                        reason: err.to_string(),
                    })
                })?;
                let timeline = Timeline::new(TimelineName::try_new(*index)?, *time_type);

                let per_timeline = per_entity.entry(entity_path.clone()).or_default();
                let per_component = per_timeline.entry(timeline).or_default();
                let per_chunk = per_component.entry(component).or_default();

                let start = col_start_raw[i];
                let end = col_end_raw[i];
                let num_rows = col_num_rows_raw[i];
                let entry = RrdManifestTemporalMapEntry {
                    time_range: AbsoluteTimeRange::new(start, end),
                    num_rows,
                };

                per_chunk
                    .entry(chunk_id)
                    .and_modify(|tr| *tr = entry)
                    .or_insert(entry);
            }
        }

        Ok(per_entity)
    }
}

// Schema fields are stored as Vecs, but we don't want their order to matter when performing comparisons.
impl PartialEq for RawRrdManifest {
    fn eq(&self, other: &Self) -> bool {
        let Self {
            store_id,
            sorbet_schema,
            sorbet_schema_sha256,
            data,
        } = self;

        *store_id == other.store_id
            && *data == other.data
            && *sorbet_schema_sha256 == other.sorbet_schema_sha256
            && sorbet_schema.metadata() == other.sorbet_schema.metadata()
            && sorbet_schema.fields.len() == other.sorbet_schema.fields.len()
            && {
                let sorted_fields = sorbet_schema.fields.iter().sorted_by_key(|f| f.name());
                let other_sorted_fields = other
                    .sorbet_schema
                    .fields
                    .iter()
                    .sorted_by_key(|f| f.name());
                itertools::izip!(sorted_fields, other_sorted_fields).all(|(f1, f2)| f1 == f2)
            }
    }
}

// Index-column helpers.
//
// Rerun index columns are tagged with `rerun:*` metadata keys that describe what kind of index
// they represent (static vs temporal, sequence vs timestamp, per-component or global). The marker
// (`start`, `end`, `num_rows`, …) is the last `:`-separated part of the column name. These
// helpers centralize the conventions so downstream code doesn't reach into names or metadata.
impl RawRrdManifest {
    /// Field metadata key holding the index name an index column belongs to.
    const FIELD_METADATA_KEY_INDEX: &str = "rerun:index";

    /// Field metadata key holding the index kind (`"sequence"`, `"timestamp"`, `"duration"`).
    const FIELD_METADATA_KEY_INDEX_KIND: &str = "rerun:index_kind";

    /// Field metadata key holding the index marker in the Segment Manifest schema, where the RRD
    /// manifest encodes it as the column name's last `:`-separated part instead.
    const FIELD_METADATA_KEY_INDEX_MARKER: &str = "rerun:index_marker";

    /// The `rerun:index` value of the static-data pseudo-index.
    ///
    /// The Segment Manifest schema tags the same columns with [`Self::INDEX_NAME_STATIC_SEGMENT_MANIFEST`] instead.
    const INDEX_NAME_STATIC: &str = "rerun:static";

    /// The Segment Manifest's `rerun:index` value of the static-data pseudo-index.
    const INDEX_NAME_STATIC_SEGMENT_MANIFEST: &str = "static";

    const INDEX_MARKER_START: &str = "start";
    const INDEX_MARKER_END: &str = "end";
    const INDEX_MARKER_NUM_ROWS: &str = "num_rows";
    const INDEX_MARKER_HAS_DATA: &str = "has_data";
    const INDEX_MARKER_HAS_STATIC_DATA: &str = "has_static_data";

    /// The Segment Manifest's global chunk-length column; the RRD manifest has no such column.
    const INDEX_MARKER_LEN: &str = "len";

    /// `true` if the field is a Rerun index column (temporal or static).
    pub fn is_index(field: &Field) -> bool {
        field
            .metadata()
            .contains_key(Self::FIELD_METADATA_KEY_INDEX)
    }

    /// The component identifier of a per-component index column, if it is one.
    pub fn get_component(field: &Field) -> Option<&str> {
        field
            .metadata()
            .get(FIELD_METADATA_KEY_COMPONENT)
            .map(|s| s.as_str())
    }

    /// The component type of a per-component index column, if its descriptor has one.
    pub fn get_component_type(field: &Field) -> Option<&str> {
        field
            .metadata()
            .get(FIELD_METADATA_KEY_COMPONENT_TYPE)
            .map(|s| s.as_str())
    }

    /// `true` if the field is a per-component index column (as opposed to a global one).
    pub fn is_index_per_component(field: &Field) -> bool {
        field.metadata().contains_key(FIELD_METADATA_KEY_COMPONENT)
    }

    /// The index name (e.g. `"frame_nr"`, `"log_time"`, `"static"`) for a field, if any.
    pub fn get_index_name(field: &Field) -> Option<&str> {
        field
            .metadata()
            .get(Self::FIELD_METADATA_KEY_INDEX)
            .map(|s| s.as_str())
    }

    /// The index kind (`"sequence"`, `"timestamp"`, `"duration"`) for a field, if any.
    pub fn get_index_kind(field: &Field) -> Option<&str> {
        field
            .metadata()
            .get(Self::FIELD_METADATA_KEY_INDEX_KIND)
            .map(|s| s.as_str())
    }

    /// `true` if the field is a Rerun index column with the given name.
    pub fn is_specific_index(field: &Field, index_name: &str) -> bool {
        Self::get_index_name(field) == Some(index_name)
    }

    /// `true` if the field belongs to the static-data pseudo-index, in either the RRD manifest or
    /// the Segment Manifest schema.
    pub fn is_index_static(field: &Field) -> bool {
        Self::get_index_name(field).is_some_and(|name| {
            name == Self::INDEX_NAME_STATIC || name == Self::INDEX_NAME_STATIC_SEGMENT_MANIFEST
        })
    }

    /// `true` if the field carries the given marker: a `rerun:index_marker` metadata entry, or
    /// the name's last `:`-separated part. The manifest writer only sets the latter; the former
    /// is what the cloud schema uses for the same columns.
    pub fn has_index_marker(field: &Field, marker: &str) -> bool {
        field
            .metadata()
            .get(Self::FIELD_METADATA_KEY_INDEX_MARKER)
            .is_some_and(|m| m == marker)
            || field
                .name()
                .rsplit_once(':')
                .is_some_and(|(_, suffix)| suffix == marker)
    }

    /// `true` if the field is the `:start` column of an index.
    pub fn is_index_start(field: &Field) -> bool {
        Self::has_index_marker(field, Self::INDEX_MARKER_START)
    }

    /// `true` if the field is the `:end` column of an index.
    pub fn is_index_end(field: &Field) -> bool {
        Self::has_index_marker(field, Self::INDEX_MARKER_END)
    }

    /// `true` if the field is the `:num_rows` column of a per-component index.
    pub fn is_index_num_rows(field: &Field) -> bool {
        Self::has_index_marker(field, Self::INDEX_MARKER_NUM_ROWS)
    }

    /// `true` if the field is the `:len` column of a Segment Manifest global index.
    pub fn is_index_length(field: &Field) -> bool {
        Self::has_index_marker(field, Self::INDEX_MARKER_LEN)
    }

    /// `true` if the field is the `:has_static_data` column of a component.
    pub fn is_index_has_static_data(field: &Field) -> bool {
        Self::has_index_marker(field, Self::INDEX_MARKER_HAS_STATIC_DATA)
    }

    /// `true` if the field is a temporal index column (not static, not per-component).
    pub fn is_index_global_temporal(field: &Field) -> bool {
        Self::is_index(field)
            && !Self::is_index_static(field)
            && !Self::is_index_per_component(field)
    }
}

// Helpers
impl RawRrdManifest {
    pub fn compute_sorbet_schema_sha256(
        schema: &arrow::datatypes::Schema,
    ) -> Result<[u8; 32], ArrowError> {
        let schema = {
            // Sort and remove top-level metadata before hashing.
            let mut fields = schema.fields().to_vec();
            fields.sort();
            arrow::datatypes::Schema::new_with_metadata(fields, Default::default()) // no metadata!
        };

        let partition_schema_ipc = {
            let mut schema_ipc = Vec::new();
            arrow::ipc::writer::StreamWriter::try_new(&mut schema_ipc, &schema)?;
            schema_ipc
        };

        use sha2::Digest as _;
        let mut hash = [0u8; 32];
        let mut hasher = sha2::Sha256::new();
        hasher.update(&partition_schema_ipc);
        hasher.finalize_into(sha2::digest::generic_array::GenericArray::from_mut_slice(
            &mut hash,
        ));

        Ok(hash)
    }

    /// Computes the appropriate column name for the provided parts.
    ///
    /// The name is guaranteed to be sanitized and safe to use in all environments where Rerun data
    /// can generally be found (Lance, external dataframe libraries, etc).
    ///
    /// If caller doesn't provide any part (i.e. all are `None`), an empty string is returned.
    pub fn compute_column_name(
        entity_path: Option<&EntityPath>,
        strip_entity_prefix: Option<&str>,
        component_desc: Option<&ComponentDescriptor>,
        prefix: Option<&str>,
        suffix: Option<&str>,
    ) -> String {
        use re_types_core::reflection::ComponentDescriptorExt as _;
        let full_name = [
            prefix.map(ToOwned::to_owned),
            entity_path.map(|p| {
                let path = p.to_string();
                // Optionally strip the entity prefix if provided
                let path = strip_entity_prefix
                    .and_then(|prefix| path.strip_prefix(prefix))
                    .unwrap_or(&path);
                // Always strip trailing slashes (if present)
                path.strip_suffix("/").unwrap_or(path).to_owned()
            }),
            component_desc
                .and_then(|descr| descr.archetype)
                .map(|archetype| archetype.short_name().to_owned()),
            component_desc.map(|descr| descr.archetype_field_name().to_owned()),
            suffix.map(ToOwned::to_owned),
        ]
        .into_iter()
        .flatten()
        .filter(|s| !s.is_empty())
        .collect::<Vec<_>>()
        .join(":");

        // All of the following characters have proven to be problematic in one or more
        // environments. Replace them with `_`, unconditionally.
        let sanitized = full_name.replace([',', ' ', '-', '.', '\\'], "_");

        // Remove leading underscore if present
        sanitized.trim_start_matches('_').to_owned()
    }

    // Prune the RecordBatch to keep only columns needed for chunk fetching.
    //
    // The full manifest may have 1000+ sparse columns (one per timeline×component pair),
    // which Arrow inflates because it allocates full-length buffers even for
    // mostly-null columns. The maps above already captured all indexing data from these
    // sparse columns; the RecordBatch is only used for `take_record_batch` → `FetchChunks`,
    // which needs exactly: chunk_id, chunk_key, chunk_partition_id, rerun_partition_layer.
    //
    // By dropping the sparse columns here, we reduce memory ~10-20x
    // and make `take_record_batch` 100x faster.
    pub(super) fn chunk_fetcher_record_batch(&self) -> RecordBatch {
        let columns_to_keep = super::RrdManifest::CHUNK_FETCHER_COLUMNS;

        let schema = self.data.schema_ref();
        let indices: Vec<usize> = schema
            .fields()
            .iter()
            .enumerate()
            .filter(|(_, field)| columns_to_keep.contains(&field.name().as_str()))
            .map(|(i, _)| i)
            .collect();

        self.data
            .project(&indices)
            .unwrap_or_else(|_| self.data.clone())
    }
}

// Sanity checks
impl RawRrdManifest {
    /// Checks the manifest for any traces of corruption.
    ///
    /// This is cheap to compute and is automatically performed when converting an [`RawRrdManifest`]
    /// from its transport-level to its application-level representation (and vice-versa).
    ///
    /// See [`Self::sanity_check_heavy`] for a more costly version that is not suitable to use in
    /// production, but can be useful in e.g. tests.
    pub fn sanity_check_cheap(&self) -> CodecResult<()> {
        re_tracing::profile_function!();
        self.check_global_columns_are_correct()?;
        self.check_index_columns_are_correct()?;
        self.check_manifest_schema_matches_sorbet_schema()?;
        Ok(())
    }

    /// Checks the manifest for any traces of corruption.
    ///
    /// This is quite costly and therefore should not be used on the happy production path.
    /// Prefer [`Self::sanity_check_cheap`] for that instead.
    pub fn sanity_check_heavy(&self) -> CodecResult<()> {
        re_tracing::profile_function!();
        self.check_sorbet_schema_sha256_is_correct()?;
        Ok(())
    }

    /// Cheap.
    fn check_global_columns_are_correct(&self) -> CodecResult<()> {
        _ = self.col_chunk_id()?;
        _ = self.col_chunk_is_static()?;
        _ = self.col_chunk_num_rows()?;
        _ = self.col_chunk_entity_path()?;
        _ = self.col_chunk_byte_size_uncompressed()?;

        // The basic size/offset columns are always there, even if they might be logically
        // superseded by a `chunk_key` column (which is backend-specific, and therefore optional).
        _ = self.col_chunk_byte_offset()?;
        _ = self.col_chunk_byte_size()?;

        if self
            .data
            .schema_ref()
            .column_with_name(Self::COLUMN_CHUNK_KEY.name)
            .is_some()
        {
            _ = self.col_chunk_key()?;
        }

        Ok(())
    }

    /// `rerun:kind` value of a Sorbet index column.
    const SORBET_KIND_INDEX: &str = "index";

    /// `rerun:kind` value of a Sorbet component-data column.
    const SORBET_KIND_DATA: &str = "data";

    /// Cheap.
    fn check_index_columns_are_correct(&self) -> CodecResult<()> {
        {
            // All columns either end in :has_static_data or :num_rows or :start or :end (or are global).
            for field in self.data.schema().fields() {
                if let Some((_, suffix)) = field.name().rsplit_once(':') {
                    match suffix {
                        Self::INDEX_MARKER_START | Self::INDEX_MARKER_END => {
                            // Checked in depth below
                        }

                        Self::INDEX_MARKER_HAS_STATIC_DATA => {
                            if *field.data_type() != Self::COLUMN_CHUNK_IS_STATIC.data_type() {
                                return Err(CodecError::from(ChunkError::Malformed {
                                    reason: format!(
                                        "field '{}' should be {} but is actually {}",
                                        field.name(),
                                        Self::COLUMN_CHUNK_IS_STATIC.data_type(),
                                        field.data_type(),
                                    ),
                                }));
                            }
                        }

                        Self::INDEX_MARKER_NUM_ROWS => {
                            if *field.data_type() != Self::COLUMN_CHUNK_NUM_ROWS.data_type() {
                                return Err(CodecError::from(ChunkError::Malformed {
                                    reason: format!(
                                        "field '{}' should be {} but is actually {}",
                                        field.name(),
                                        Self::COLUMN_CHUNK_NUM_ROWS.data_type(),
                                        field.data_type(),
                                    ),
                                }));
                            }
                        }

                        suffix => {
                            return Err(CodecError::from(ChunkError::Malformed {
                                reason: format!(
                                    "field '{}' has invalid suffix '{suffix}'",
                                    field.name(),
                                ),
                            }));
                        }
                    }
                } else {
                    // Global column
                    match field.name().as_str() {
                        name if Self::GLOBAL_COLUMN_NAMES.contains(&name) => {}

                        name if Self::COMMON_IMPL_SPECIFIC_FIELDS.contains(&name) => {}

                        name => {
                            return Err(CodecError::from(ChunkError::Malformed {
                                reason: format!(
                                    "unexpected field '{name}' should not be present in an RRD manifest",
                                ),
                            }));
                        }
                    }
                }
            }
        }

        {
            // All `:start` columns should have a matching `:end`.
            // All `:end` columns should have a matching `:start`.
            for field in self.data.schema().fields() {
                if let Some((prefix, suffix)) = field.name().rsplit_once(':') {
                    let counterpart = match suffix {
                        "start" => "end",
                        "end" => "start",
                        _ => continue,
                    };

                    let field_counterpart = self
                        .data
                        .schema_ref()
                        .field_with_name(&format!("{prefix}:{counterpart}"))
                        .map_err(|_err| {
                            CodecError::from(ChunkError::Malformed {
                                reason: format!(
                                    "field '{}' does not have matching `:{counterpart}` field",
                                    field.name()
                                ),
                            })
                        })?;

                    match field.data_type() {
                        arrow::datatypes::DataType::Int64
                        | arrow::datatypes::DataType::Timestamp(_, _)
                        | arrow::datatypes::DataType::Duration(_) => {}

                        datatype => {
                            return Err(CodecError::from(ChunkError::Malformed {
                                reason: format!(
                                    "field '{}' is {datatype} which is not a supported index datatype",
                                    field.name(),
                                ),
                            }));
                        }
                    }

                    if field.data_type() != field_counterpart.data_type() {
                        return Err(CodecError::from(ChunkError::Malformed {
                            reason: format!(
                                "field '{}' is {} but field '{}' is {}",
                                field.name(),
                                field.data_type(),
                                field_counterpart.name(),
                                field_counterpart.data_type()
                            ),
                        }));
                    }
                }
            }
        }

        {
            // All `:start` columns should have a matching `:num_rows`.
            for field in self.data.schema().fields() {
                if let Some((prefix, "num_rows")) = field.name().rsplit_once(':') {
                    let field_num_rows = self
                        .data
                        .schema_ref()
                        .field_with_name(&format!("{prefix}:num_rows"))
                        .map_err(|_err| {
                            CodecError::from(ChunkError::Malformed {
                                reason: format!(
                                    "field '{}' does not have matching `:num_rows` field",
                                    field.name()
                                ),
                            })
                        })?;

                    match field_num_rows.data_type() {
                        arrow::datatypes::DataType::UInt64 => {}
                        datatype => {
                            return Err(CodecError::from(ChunkError::Malformed {
                                reason: format!(
                                    "field '{}' is {datatype} while it should be UInt64Array",
                                    field_num_rows.name(),
                                ),
                            }));
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Cheap.
    fn check_manifest_schema_matches_sorbet_schema(&self) -> CodecResult<()> {
        let any_static_chunks = self.col_chunk_is_static_iter()?.any(|b| b);

        let sorbet_indexes = self
            .sorbet_schema
            .fields()
            .iter()
            .filter_map(|f| {
                let md = f.metadata();
                (md.get(re_sorbet::RERUN_KIND).map(|s| s.as_str()) == Some(Self::SORBET_KIND_INDEX))
                    .then(|| md.contains_key(re_sorbet::SORBET_INDEX_NAME).then_some(f))
                    .flatten()
            })
            .unique()
            .collect_vec();

        let sorbet_columns = self
            .sorbet_schema
            .fields()
            .iter()
            .filter(|f| {
                f.metadata().get(re_sorbet::RERUN_KIND).map(|s| s.as_str())
                    == Some(Self::SORBET_KIND_DATA)
            })
            .unique()
            .collect_vec();

        if any_static_chunks {
            // If there are any static chunks, then all components must have :has_static_data indexes.
            for column in &sorbet_columns {
                let md = column.metadata();
                let Some(component) = md.get(FIELD_METADATA_KEY_COMPONENT) else {
                    return Err(CodecError::from(ChunkError::Malformed {
                        reason: format!(
                            "column '{}' is missing rerun:component metadata",
                            column.name()
                        ),
                    }));
                };
                let descr = ComponentDescriptor {
                    archetype: md
                        .get(FIELD_METADATA_KEY_ARCHETYPE)
                        .and_then(|s| ArchetypeName::try_new(s).ok()),
                    component: ComponentIdentifier::try_new(component).map_err(|err| {
                        CodecError::from(ChunkError::Malformed {
                            reason: err.to_string(),
                        })
                    })?,
                    component_type: md
                        .get(FIELD_METADATA_KEY_COMPONENT_TYPE)
                        .and_then(|s| ComponentType::try_new(s).ok()),
                };
                let column_name = Self::compute_column_name(
                    None,
                    None,
                    Some(&descr),
                    None,
                    Some(Self::INDEX_MARKER_HAS_STATIC_DATA),
                );

                self.data
                    .schema_ref()
                    .field_with_name(&column_name)
                    .map_err(|_err| {
                        CodecError::from(ChunkError::Malformed {
                            reason: format!("static index '{column_name}' is missing"),
                        })
                    })?;
            }
        }

        let mut rrd_manifest_fields: HashMap<_, _> = self
            .data
            .schema_ref()
            .fields()
            .iter()
            .filter(|f| Self::is_index_start(f) || Self::is_index_end(f))
            .map(|f| (f.name(), f))
            .collect();

        for sorbet_index in &sorbet_indexes {
            // We must account for the fact that names in the Sorbet schema are not normalized yet.
            let sorbet_index_name_normalized =
                Self::compute_column_name(None, None, None, Some(sorbet_index.name()), None);

            // All global indexes should have :start and :end columns of the right type.
            for suffix in [Self::INDEX_MARKER_START, Self::INDEX_MARKER_END] {
                let field = rrd_manifest_fields.remove(&format!("{sorbet_index_name_normalized}:{suffix}"))
                    .ok_or_else(|| {
                        CodecError::from(ChunkError::Malformed {
                            reason: format!(
                                "global index '{sorbet_index}' does not have matching `:{suffix}` field"
                            ),
                        })
                    })?;

                if sorbet_index.data_type() != field.data_type() {
                    return Err(CodecError::from(ChunkError::Malformed {
                        reason: format!(
                            "global index '{}' is {} but '{}' is {}",
                            sorbet_index.name(),
                            sorbet_index.data_type(),
                            field.name(),
                            field.data_type()
                        ),
                    }));
                }
            }

            // All local indexes should exist and have :start and :end columns of the right type.
            for sorbet_column in &sorbet_columns {
                let md = sorbet_column.metadata();

                let Some(component) = md.get(FIELD_METADATA_KEY_COMPONENT) else {
                    return Err(CodecError::from(ChunkError::Malformed {
                        reason: format!(
                            "column '{}' is missing rerun:component metadata",
                            sorbet_column.name()
                        ),
                    }));
                };
                let descr = ComponentDescriptor {
                    archetype: md
                        .get(FIELD_METADATA_KEY_ARCHETYPE)
                        .and_then(|s| ArchetypeName::try_new(s).ok()),
                    component: ComponentIdentifier::try_new(component).map_err(|err| {
                        CodecError::from(ChunkError::Malformed {
                            reason: err.to_string(),
                        })
                    })?,
                    component_type: md
                        .get(FIELD_METADATA_KEY_COMPONENT_TYPE)
                        .and_then(|s| ComponentType::try_new(s).ok()),
                };

                for suffix in [Self::INDEX_MARKER_START, Self::INDEX_MARKER_END] {
                    let column_name = Self::compute_column_name(
                        None,
                        None,
                        Some(&descr),
                        Some(sorbet_index.name()),
                        Some(suffix),
                    );

                    if md.get(re_sorbet::SORBET_IS_STATIC).map(|s| s.as_str()) == Some("true") {
                        // Static columns don't have :start nor :end columns… unless they exist
                        // both temporally and statically, something which is legal in Rerun, and
                        // will end up with a final Sorbet schema that declares those column as
                        // static (which is correct, since static overrides everything else), even
                        // though there are still traces of temporal data for that same column!
                        _ = rrd_manifest_fields.remove(&column_name);
                        continue;
                    }

                    let Some(field) = rrd_manifest_fields.remove(&column_name) else {
                        // Not all indexes have all components, that's fine.
                        continue;
                    };

                    if sorbet_index.data_type() != field.data_type() {
                        return Err(CodecError::from(ChunkError::Malformed {
                            reason: format!(
                                "local index '{}' is {} but '{}' is {}",
                                sorbet_index.name(),
                                sorbet_index.data_type(),
                                field.name(),
                                field.data_type()
                            ),
                        }));
                    }
                }
            }
        }

        if !rrd_manifest_fields.is_empty() {
            return Err(CodecError::from(ChunkError::Malformed {
                reason: format!(
                    "detected dangling indexes (present in manifest but not in Sorbet schema): {:?}",
                    rrd_manifest_fields.keys()
                ),
            }));
        }

        Ok(())
    }

    /// Costly.
    fn check_sorbet_schema_sha256_is_correct(&self) -> CodecResult<()> {
        let expected_sorbet_schema_sha256 = Self::compute_sorbet_schema_sha256(&self.sorbet_schema)
            .map_err(CodecError::ArrowDeserialization)?;

        if self.sorbet_schema_sha256 != expected_sorbet_schema_sha256 {
            return Err(CodecError::ArrowDeserialization(ArrowError::SchemaError(
                format!(
                    "invalid schema hash: expected {} but got {}",
                    sha256_to_hex(&expected_sorbet_schema_sha256),
                    sha256_to_hex(&self.sorbet_schema_sha256),
                ),
            )));
        }
        Ok(())
    }
}

// Fields
impl RawRrdManifest {
    /// The ID of the chunk. Every chunk has one.
    pub const COLUMN_CHUNK_ID: quiver::ColumnDesc<ChunkId> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_id");

    /// Does the chunk hold static data? Every chunk is either static or temporal.
    pub const COLUMN_CHUNK_IS_STATIC: quiver::ColumnDesc<bool> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_is_static");

    /// How many rows the chunk holds. Every chunk has a row count.
    pub const COLUMN_CHUNK_NUM_ROWS: quiver::ColumnDesc<u64> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_num_rows");

    /// The entity path of the chunk. Every chunk has one.
    pub const COLUMN_CHUNK_ENTITY_PATH: quiver::ColumnDesc<EntityPath> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_entity_path");

    /// Where the chunk's payload starts.
    ///
    /// See the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    pub const COLUMN_CHUNK_BYTE_OFFSET: quiver::ColumnDesc<u64> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_byte_offset");

    /// How long the chunk's payload is.
    ///
    /// See the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    pub const COLUMN_CHUNK_BYTE_SIZE: quiver::ColumnDesc<u64> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_byte_size");

    /// How long the chunk's payload would be, uncompressed.
    ///
    /// See the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    pub const COLUMN_CHUNK_BYTE_SIZE_UNCOMPRESSED: quiver::ColumnDesc<u64> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_byte_size_uncompressed");

    /// Opaque key encoding where to fetch the chunk. Every chunk has one.
    pub const COLUMN_CHUNK_KEY: quiver::ColumnDesc<quiver::Binary> =
        quiver::ColumnDesc::new("RawRrdManifest", "chunk_key");

    /// The names of every global (i.e. non-index) column of an RRD manifest.
    ///
    /// Index columns are named after the timeline and component they describe,
    /// so they cannot be listed here (see [`Self::compute_column_name`]).
    pub const GLOBAL_COLUMN_NAMES: &[&str] = &[
        Self::COLUMN_CHUNK_ID.name,
        Self::COLUMN_CHUNK_IS_STATIC.name,
        Self::COLUMN_CHUNK_NUM_ROWS.name,
        Self::COLUMN_CHUNK_ENTITY_PATH.name,
        Self::COLUMN_CHUNK_BYTE_OFFSET.name,
        Self::COLUMN_CHUNK_BYTE_SIZE.name,
        Self::COLUMN_CHUNK_BYTE_SIZE_UNCOMPRESSED.name,
        Self::COLUMN_CHUNK_KEY.name,
    ];

    /// These fields might be returned by some implementations (such as Rerun Hub) that do not
    /// support fetching chunks with only a set of chunk-keys.
    /// We generally want to ignore them during tests and sanity checking, and just blindly forward
    /// them as-is otherwise.
    pub const COMMON_IMPL_SPECIFIC_FIELDS: &[&str] = &[
        "chunk_partition_id",
        "chunk_partition_layer",
        "rerun_partition_id",
        "rerun_partition_layer",
        "chunk_segment_id",
        "chunk_segment_layer",
        "rerun_segment_id",
        "rerun_segment_layer",
    ];

    pub fn field_index_start(timeline: &Timeline, desc: Option<&ComponentDescriptor>) -> Field {
        Self::any_index_field(
            timeline,
            timeline.datatype(),
            desc,
            Self::INDEX_MARKER_START,
        )
    }

    pub fn field_index_end(timeline: &Timeline, desc: Option<&ComponentDescriptor>) -> Field {
        Self::any_index_field(timeline, timeline.datatype(), desc, Self::INDEX_MARKER_END)
    }

    pub fn field_index_num_rows(timeline: &Timeline, desc: Option<&ComponentDescriptor>) -> Field {
        Self::any_index_field(
            timeline,
            arrow::datatypes::DataType::UInt64,
            desc,
            Self::INDEX_MARKER_NUM_ROWS,
        )
    }

    pub fn field_index_has_data(timeline: &Timeline, desc: &ComponentDescriptor) -> Field {
        Self::any_index_field(
            timeline,
            arrow::datatypes::DataType::Boolean,
            Some(desc),
            Self::INDEX_MARKER_HAS_DATA,
        )
    }

    pub fn field_has_static_data(desc: &ComponentDescriptor) -> Field {
        let field_name = Self::compute_column_name(
            None,
            None,
            Some(desc),
            None,
            Some(Self::INDEX_MARKER_HAS_STATIC_DATA),
        );

        let mut metadata = std::collections::HashMap::default();
        metadata.extend(
            [
                Some((
                    Self::FIELD_METADATA_KEY_INDEX.to_owned(),
                    Self::INDEX_NAME_STATIC.to_owned(),
                )),
                desc.component_type.map(|component_type| {
                    (
                        FIELD_METADATA_KEY_COMPONENT_TYPE.to_owned(),
                        component_type.full_name().to_owned(),
                    )
                }),
                desc.archetype.as_ref().map(|name| {
                    (
                        FIELD_METADATA_KEY_ARCHETYPE.to_owned(),
                        name.full_name().to_owned(),
                    )
                }),
                Some((
                    FIELD_METADATA_KEY_COMPONENT.to_owned(),
                    desc.component.to_string(),
                )),
            ]
            .into_iter()
            .flatten(),
        );

        let nullable = false;
        Field::new(field_name, arrow::datatypes::DataType::Boolean, nullable)
            .with_metadata(metadata)
    }

    // ---

    fn any_index_field(
        timeline: &Timeline,
        datatype: arrow::datatypes::DataType,
        desc: Option<&ComponentDescriptor>,
        marker: &str,
    ) -> Field {
        let index_name = timeline.name();

        let field_name =
            Self::compute_column_name(None, None, desc, Some(index_name), Some(marker));

        let mut metadata = std::collections::HashMap::default();
        metadata.extend([(
            Self::FIELD_METADATA_KEY_INDEX.to_owned(),
            timeline.name().to_string(),
        )]);
        if let Some(desc) = desc {
            metadata.extend(
                [
                    desc.component_type.map(|component_type| {
                        (
                            FIELD_METADATA_KEY_COMPONENT_TYPE.to_owned(),
                            component_type.full_name().to_owned(),
                        )
                    }),
                    desc.archetype.as_ref().map(|name| {
                        (
                            FIELD_METADATA_KEY_ARCHETYPE.to_owned(),
                            name.full_name().to_owned(),
                        )
                    }),
                    Some((
                        FIELD_METADATA_KEY_COMPONENT.to_owned(),
                        desc.component.to_string(),
                    )),
                ]
                .into_iter()
                .flatten(),
            );
        }

        let nullable = true; // A) static B) not all chunks belong to all timelines
        Field::new(field_name, datatype, nullable).with_metadata(metadata)
    }
}

// Column accessors
impl RawRrdManifest {
    /// The entity path column.
    pub fn col_chunk_entity_path(&self) -> CodecResult<quiver::Column<EntityPath>> {
        Ok(Self::COLUMN_CHUNK_ENTITY_PATH.extract(&self.data)?)
    }

    /// Returns an iterator over the decoded Arrow data for the entity path column.
    ///
    /// This might incur interning costs, but is otherwise basically free.
    pub fn col_chunk_entity_path_iter(&self) -> CodecResult<impl Iterator<Item = EntityPath>> {
        Ok(self.col_chunk_entity_path()?.into_iter_owned())
    }

    /// The chunk ID column.
    pub fn col_chunk_id(&self) -> CodecResult<quiver::Column<ChunkId>> {
        Ok(Self::COLUMN_CHUNK_ID.extract(&self.data)?)
    }

    /// Returns an iterator over the decoded Arrow data for the chunk ID column.
    ///
    /// This is free.
    pub fn col_chunk_id_iter(&self) -> CodecResult<impl Iterator<Item = ChunkId>> {
        Ok(self.col_chunk_id()?.into_iter_owned())
    }

    /// The is-static column.
    pub fn col_chunk_is_static(&self) -> CodecResult<quiver::Column<bool>> {
        Ok(Self::COLUMN_CHUNK_IS_STATIC.extract(&self.data)?)
    }

    /// Returns an iterator over the decoded Arrow data for the is-static column.
    ///
    /// This is free.
    pub fn col_chunk_is_static_iter(&self) -> CodecResult<impl Iterator<Item = bool>> {
        Ok(self.col_chunk_is_static()?.into_iter_owned())
    }

    /// The num-rows column.
    pub fn col_chunk_num_rows(&self) -> CodecResult<quiver::Column<u64>> {
        Ok(Self::COLUMN_CHUNK_NUM_ROWS.extract(&self.data)?)
    }

    /// Returns an iterator over the decoded Arrow data for the num-rows column.
    ///
    /// This is free.
    pub fn col_chunk_num_rows_iter(&self) -> CodecResult<impl Iterator<Item = u64>> {
        Ok(self.col_chunk_num_rows()?.into_iter_owned())
    }

    /// The byte-offset column.
    ///
    /// See also the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    pub fn col_chunk_byte_offset(&self) -> CodecResult<quiver::Column<u64>> {
        Ok(Self::COLUMN_CHUNK_BYTE_OFFSET.extract(&self.data)?)
    }

    /// Returns an iterator over the decoded Arrow data for the byte-offset column.
    ///
    /// This is free.
    pub fn col_chunk_byte_offset_iter(&self) -> CodecResult<impl Iterator<Item = u64>> {
        Ok(self.col_chunk_byte_offset()?.into_iter_owned())
    }

    /// The byte-size column.
    ///
    /// See also the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    pub fn col_chunk_byte_size(&self) -> CodecResult<quiver::Column<u64>> {
        Ok(Self::COLUMN_CHUNK_BYTE_SIZE.extract(&self.data)?)
    }

    /// Returns an iterator over the decoded Arrow data for the byte-size column.
    ///
    /// See also the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    ///
    /// This is free.
    pub fn col_chunk_byte_size_iter(&self) -> CodecResult<impl Iterator<Item = u64>> {
        Ok(self.col_chunk_byte_size()?.into_iter_owned())
    }

    /// The *uncompressed* byte-size column.
    ///
    /// See also the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    pub fn col_chunk_byte_size_uncompressed(&self) -> CodecResult<quiver::Column<u64>> {
        Ok(Self::COLUMN_CHUNK_BYTE_SIZE_UNCOMPRESSED.extract(&self.data)?)
    }

    /// Returns an iterator over the decoded Arrow data for the *uncompressed* byte-size column.
    ///
    /// See also the `Understand size/offset columns` section of the [`RawRrdManifest`] documentation.
    ///
    /// This is free.
    pub fn col_chunk_byte_size_uncompressed_iter(&self) -> CodecResult<impl Iterator<Item = u64>> {
        Ok(self.col_chunk_byte_size_uncompressed()?.into_iter_owned())
    }

    /// The chunk-key column, if present.
    ///
    /// Read as optional: a merged manifest can have nulls here, since concatenating a manifest
    /// that has chunk keys with one that does not leaves the rows of the latter null (see
    /// `RrdManifest::add_null_chunk_key_column`).
    pub fn col_chunk_key(&self) -> CodecResult<quiver::Column<Option<quiver::Binary>>> {
        Ok(Self::COLUMN_CHUNK_KEY.optional().extract(&self.data)?)
    }
}

/// Strip the null mask from columns that the [`RawRrdManifest`] contract says are non-nullable
/// with a default value (`:has_static_data` → `false`, `:num_rows` → `0`).
///
/// After [`crate::RrdManifestBuilder`] emits a manifest for a single layer, these columns are
/// fully populated. But once we merge multiple per-layer manifests via polymorphic concat, a
/// component only known to one layer will have nulls in the rows from other layers. The contract
/// — shared with commercial (see `ExtendedRrdManifest::try_into_standard_manifest`) — is that
/// the output manifest has these columns non-nullable with the underlying buffer's zero bits
/// (which Arrow zero-initializes when `new_null_array` is called).
fn strip_null_mask_on_default_columns(data: RecordBatch) -> CodecResult<RecordBatch> {
    use re_arrow_util::ArrowArrayDowncastRef as _;

    let (schema, mut columns, num_rows) = data.into_parts();
    let mut new_fields = schema.fields.to_vec();

    for (field, column) in itertools::izip!(&mut new_fields, &mut columns) {
        if column.null_count() == 0 {
            continue;
        }

        let name = field.name().as_str();
        if RawRrdManifest::is_index_has_static_data(field) {
            let Some(c) = column.downcast_array_ref::<BooleanArray>() else {
                return Err(CodecError::ArrowDeserialization(ArrowError::SchemaError(
                    format!(
                        "'{name}' should be a BooleanArray, got {}",
                        column.data_type()
                    ),
                )));
            };
            let (bools, _nulls) = c.clone().into_parts();
            *column = std::sync::Arc::new(BooleanArray::new(bools, None));
            *field = std::sync::Arc::new((**field).clone().with_nullable(false));
        } else if RawRrdManifest::is_index_num_rows(field) {
            let Some(c) = column.downcast_array_ref::<UInt64Array>() else {
                return Err(CodecError::ArrowDeserialization(ArrowError::SchemaError(
                    format!(
                        "'{name}' should be a UInt64Array, got {}",
                        column.data_type()
                    ),
                )));
            };
            let (_dt, ints, _nulls) = c.clone().into_parts();
            *column = std::sync::Arc::new(UInt64Array::new(ints, None));
            *field = std::sync::Arc::new((**field).clone().with_nullable(false));
        }
    }

    let schema = arrow::datatypes::Schema::new_with_metadata(new_fields, schema.metadata.clone());
    RecordBatch::try_new_with_options(
        std::sync::Arc::new(schema),
        columns,
        &arrow::array::RecordBatchOptions::new().with_row_count(Some(num_rows)),
    )
    .map_err(CodecError::ArrowSerialization)
}