// Copyright (c) 2023-2026 ParadeDB, Inc.
//
// This file is part of ParadeDB - Postgres for Search and Analytics
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

use std::convert::identity;
use std::sync::{Arc, OnceLock};

use crate::index::reader::index::SearchIndexReader;
use crate::postgres::datetime::PostgresDateTime;
use crate::postgres::pdb_owned_value::PdbOwnedValue;
use crate::postgres::types::{is_pgoid_datetime_type, TantivyValue};
use crate::postgres::types_arrow::datetime_to_pg_micros;
use crate::scan::deferred_encode::unpack_doc_address;
use crate::schema::SearchFieldType;

use arrow_array::builder::{BinaryViewBuilder, StringViewBuilder};
use arrow_array::builder::{
    BooleanBuilder, Float64Builder, Int64Builder, TimestampMicrosecondBuilder, UInt64Builder,
};
use arrow_array::{ArrayRef, UInt64Array};
use arrow_buffer::Buffer;
use datafusion::common::Result;
use datafusion::error::DataFusionError;
use serde::{Deserialize, Serialize};
use tantivy::columnar::{BytesColumn, StrColumn};
use tantivy::fastfield::{Column, FastFieldReaders};
use tantivy::termdict::TermOrdinal;
use tantivy::{DocAddress, DocId, Searcher, SegmentOrdinal};

/// A fast-field index position value.
pub type FFIndex = usize;

/// Uniquely identifies a specific dictionary/array in the underlying storage layer
/// by combining the table's index relation ID and the fast field index.
#[derive(
    Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize,
)]
pub struct CanonicalColumn {
    pub indexrelid: u32,
    pub ff_index: FFIndex,
}

struct SegmentCache {
    columns: Vec<OnceLock<FFType>>,
    ctid: OnceLock<FFType>,
}

/// A helper for tracking specific "fast field" readers from a [`SearchIndexReader`] reference
///
/// They're organized by index positions and not names to eliminate as much runtime overhead as
/// possible when looking up the value of a specific fast field.
#[derive(Default)]
pub struct FFHelper(
    // `None` only for the `empty()`/`Default` placeholders, which are never used for column
    // access; `with_fields` always builds the full inner state.
    Option<FFInner>,
);

struct FFInner {
    // A segment's fast-field file opens on first column access: the tantivy fork's
    // `SegmentReader::fast_fields()` is a lazy open, and for a mutable segment it materializes
    // the segment from the heap. The Searcher also keeps the reader's `MVCCDirectory` and its
    // segment pins alive. Initialize on the backend thread only.
    searcher: Searcher,
    columns: Vec<WhichFastField>,
    segment_caches: Vec<SegmentCache>,
}

impl FFHelper {
    pub fn empty() -> Self {
        Self::default()
    }

    pub fn with_fields(reader: &SearchIndexReader, fields: &[WhichFastField]) -> Self {
        Self(Some(FFInner {
            searcher: reader.searcher().clone(),
            segment_caches: Self::segment_caches(reader.segment_readers().len(), fields.len()),
            columns: fields.to_vec(),
        }))
    }

    fn segment_caches(segment_count: usize, width: usize) -> Vec<SegmentCache> {
        (0..segment_count)
            .map(|_| SegmentCache {
                columns: (0..width).map(|_| OnceLock::new()).collect(),
                ctid: OnceLock::new(),
            })
            .collect()
    }

    fn inner(&self) -> &FFInner {
        self.0
            .as_ref()
            .expect("FFHelper::empty() must not be used for column access")
    }

    fn searcher(&self) -> &Searcher {
        &self.inner().searcher
    }

    fn caches(&self) -> &[SegmentCache] {
        &self.inner().segment_caches
    }

    fn fast_fields(&self, segment_ord: SegmentOrdinal) -> &FastFieldReaders {
        self.searcher().segment_reader(segment_ord).fast_fields()
    }

    pub fn ctid(&self, segment_ord: SegmentOrdinal) -> &FFType {
        self.caches()[segment_ord as usize]
            .ctid
            .get_or_init(|| FFType::new_ctid(self.fast_fields(segment_ord)))
    }

    pub fn column(&self, segment_ord: SegmentOrdinal, field: FFIndex) -> &FFType {
        self.caches()[segment_ord as usize].columns[field].get_or_init(|| {
            match &self.inner().columns[field] {
                WhichFastField::Named(name, _) | WhichFastField::Deferred(name, _) => {
                    FFType::new(self.fast_fields(segment_ord), name)
                }
                WhichFastField::Ctid
                | WhichFastField::TableOid
                | WhichFastField::Score
                | WhichFastField::Junk(_)
                | WhichFastField::DeferredCtid(_)
                | WhichFastField::MatchTag(_) => FFType::Junk,
            }
        })
    }

    #[track_caller]
    pub fn value(&self, field: FFIndex, doc_address: DocAddress) -> Option<TantivyValue> {
        Some(self.column(doc_address.segment_ord, field).value(
            doc_address.doc_id,
            self.inner().columns[field].field_type().copied(),
        ))
    }

    pub fn num_segments(&self) -> usize {
        self.0
            .as_ref()
            .map_or(0, |inner| inner.segment_caches.len())
    }
}

/// A macro to fetch values for the given ids into an Arrow array.
macro_rules! fetch_ff_column {
    ($col:expr, $ids:ident, $($ff_type:ident => $conversion:ident => $builder:ident),* $(,)?) => {
        match $col {
            $(
                FFType::$ff_type(col) => {
                    let mut column_results = Vec::with_capacity($ids.len());
                    column_results.resize($ids.len(), None);
                    col.first_vals($ids, &mut column_results);
                    let mut builder = $builder::with_capacity($ids.len());
                    for maybe_val in column_results {
                        if let Some(val) = maybe_val {
                            builder.append_value($conversion(val));
                        } else {
                            builder.append_null();
                        }
                    }
                    Arc::new(builder.finish()) as ArrayRef
                }
            )*
            x => panic!("Unhandled column type {x:?}"),
        }
    };
}

/// A macro to deduplicate fetching term ordinals for String/Bytes columns.
macro_rules! fetch_term_ords {
    ($ords:expr, $ids:expr) => {{
        let mut term_ords = Vec::with_capacity($ids.len());
        term_ords.resize($ids.len(), None);
        $ords.first_vals($ids, &mut term_ords);
        let mut builder = UInt64Builder::with_capacity($ids.len());
        for maybe_ord in term_ords {
            if let Some(ord) = maybe_ord {
                builder.append_value(ord);
            } else {
                builder.append_null();
            }
        }
        Arc::new(builder.finish()) as ArrayRef
    }};
}

/// Helper for working with different "fast field" types as if they're all one type.
///
/// This enum is used *after* a column is open to provide a typed wrapper around the underlying
/// Tantivy column readers.
#[derive(Debug)]
pub enum FFType {
    Junk,
    Text(StrColumn),
    Bytes(BytesColumn),
    I64(Column<i64>),
    F64(Column<f64>),
    U64(Column<u64>),
    Bool(Column<bool>),
    Date(Column<tantivy::DateTime>),
}

impl FFType {
    /// Construct the proper [`FFType`] for the internal `ctid` field, which
    /// should be a known field name in the Tantivy index
    pub fn new_ctid(ffr: &FastFieldReaders) -> Self {
        Self::U64(ffr.u64("ctid").expect("ctid should be a u64 fast field"))
    }

    /// Construct the proper [`FFType`] for the specified `field_name`, which
    /// should be a known field name in the Tantivy index
    #[track_caller]
    pub fn new(ffr: &FastFieldReaders, field_name: &str) -> Self {
        if let Ok(ff) = ffr.i64(field_name) {
            Self::I64(ff)
        } else if let Ok(Some(ff)) = ffr.str(field_name) {
            Self::Text(ff)
        } else if let Ok(Some(ff)) = ffr.bytes(field_name) {
            Self::Bytes(ff)
        } else if let Ok(ff) = ffr.u64(field_name) {
            Self::U64(ff)
        } else if let Ok(ff) = ffr.f64(field_name) {
            Self::F64(ff)
        } else if let Ok(ff) = ffr.bool(field_name) {
            Self::Bool(ff)
        } else if let Ok(ff) = ffr.date(field_name) {
            Self::Date(ff)
        } else {
            panic!("`{field_name}` is missing or is not configured as a fast field")
        }
    }

    /// Given a [`DocId`], what is its "fast field" value?
    #[inline(always)]
    pub fn value(&self, doc: DocId, search_field_type: Option<SearchFieldType>) -> TantivyValue {
        match self {
            FFType::Junk => TantivyValue(PdbOwnedValue::Null),
            FFType::Text(ff) => {
                let mut s = String::new();
                let ord = ff
                    .term_ords(doc)
                    .next()
                    .expect("term ord should be retrievable");
                ff.ord_to_str(ord, &mut s)
                    .expect("string should be retrievable for term ord");
                TantivyValue(s.into())
            }
            FFType::Bytes(ff) => {
                let mut bytes = Vec::new();
                let ord = ff
                    .term_ords(doc)
                    .next()
                    .expect("term ord should be retrievable");
                ff.ord_to_bytes(ord, &mut bytes)
                    .expect("bytes should be retrievable for term ord");
                TantivyValue(PdbOwnedValue::Bytes(bytes))
            }
            FFType::I64(ff) => {
                // versions >= DATETIME_I64_STORAGE_VERSION store datetimes as I64
                if let Some(sft) = search_field_type {
                    if is_pgoid_datetime_type(sft.typeoid()) {
                        let value = ff.first(doc).map(|first| {
                            let pgdt = PostgresDateTime::try_from_raw(first)
                                .expect("This should always be a valid datetime value");
                            PdbOwnedValue::Date(pgdt)
                        });
                        return TantivyValue(value.unwrap_or(PdbOwnedValue::Null));
                    }
                }
                let value = ff.first(doc).map(|first| first.into());
                TantivyValue(value.unwrap_or(PdbOwnedValue::Null))
            }
            FFType::F64(ff) => TantivyValue(
                ff.first(doc)
                    .map(|first| first.into())
                    .unwrap_or(PdbOwnedValue::Null),
            ),
            FFType::U64(ff) => TantivyValue(
                ff.first(doc)
                    .map(|first| first.into())
                    .unwrap_or(PdbOwnedValue::Null),
            ),
            FFType::Bool(ff) => TantivyValue(
                ff.first(doc)
                    .map(|first| first.into())
                    .unwrap_or(PdbOwnedValue::Null),
            ),
            FFType::Date(ff) => TantivyValue(
                ff.first(doc)
                    .map(|first| first.into())
                    .unwrap_or(PdbOwnedValue::Null),
            ),
        }
    }

    /// Given a [`DocId`], what is its u64 "fast field" value?
    ///
    /// If this [`FFType`] isn't [`FFType::U64`], this function returns [`None`].
    #[inline(always)]
    pub fn as_u64(&self, doc: DocId) -> Option<u64> {
        if let FFType::U64(ff) = self {
            ff.first(doc)
        } else {
            None
        }
    }

    /// Given [`DocId`]s, what are their u64 "fast field" values?
    ///
    /// The given `output` slice must be the same length as the docs slice.
    #[inline(always)]
    pub fn as_u64s(&self, docs: &[DocId], output: &mut [Option<u64>]) {
        let FFType::U64(ff) = self else {
            panic!("Expected a u64 column.");
        };
        ff.first_vals(docs, output);
    }

    /// Fetches the batch of fast field values (or term ordinals for Text/Bytes)
    /// as an Arrow array.
    pub fn fetch_values_or_ords_to_arrow(
        &self,
        ids: &[u32],
        search_field_type: SearchFieldType,
    ) -> ArrayRef {
        match self {
            FFType::Text(col) => fetch_term_ords!(col.ords(), ids),
            FFType::Bytes(col) => fetch_term_ords!(col.ords(), ids),
            FFType::Junk => Arc::new(arrow_array::new_null_array(
                &arrow_schema::DataType::Null,
                ids.len(),
            )),
            FFType::I64(_) if is_pgoid_datetime_type(search_field_type.typeoid()) => {
                fetch_ff_column!(self, ids, I64 => identity => TimestampMicrosecondBuilder)
            }
            numeric_column => fetch_ff_column!(numeric_column, ids,
                I64  => identity => Int64Builder,
                F64  => identity => Float64Builder,
                U64  => identity => UInt64Builder,
                Bool => identity => BooleanBuilder,
                Date => datetime_to_pg_micros => TimestampMicrosecondBuilder,
            ),
        }
    }
}

/// A request for a specific fast field, used *before* the column is open.
///
/// This enum allows consumers to specify which columns to retrieve and their expected types.
///
/// # Type Widening
///
/// Currently, we "widen" various Postgres types into larger underlying storage types (e.g.
/// based on how they are stored in Tantivy). For instance, JSON and UUID are both stored as Strings.
/// The consumer of the data (e.g. the Arrow conversion layer) is responsible for interpreting
/// these widened types back into their original Postgres OIDs via `SearchFieldType::typeoid()`.
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, Hash)]
pub enum WhichFastField {
    Junk(String),
    Ctid,
    TableOid,
    Score,
    Named(String, SearchFieldType),
    Deferred(String, SearchFieldType),
    /// Packed DocAddress ctid for deferred visibility (joinscan path only).
    /// The String is the ctid column alias (e.g. "ctid_0").
    DeferredCtid(String),
    /// Synthetic match tag column (e.g. "__users_tag_0") for disjunctive search join filtering.
    MatchTag(String),
}

impl<S: AsRef<str>> From<(S, SearchFieldType)> for WhichFastField {
    fn from(value: (S, SearchFieldType)) -> Self {
        let name = value.0.as_ref();
        match name {
            "ctid" => WhichFastField::Ctid,
            "tableoid" => WhichFastField::TableOid,
            "pdb.score()" => WhichFastField::Score,
            other => {
                if other.starts_with("junk(") && other.ends_with(")") {
                    WhichFastField::Junk(String::from(
                        other.trim_start_matches("junk(").trim_end_matches(")"),
                    ))
                } else {
                    WhichFastField::Named(String::from(other), value.1)
                }
            }
        }
    }
}

impl WhichFastField {
    pub fn name(&self) -> String {
        match self {
            WhichFastField::Junk(s) => format!("junk({s})"),
            WhichFastField::Ctid => "ctid".into(),
            WhichFastField::TableOid => "tableoid".into(),
            WhichFastField::Score => "pdb.score()".into(),
            WhichFastField::Named(s, _) => s.clone(),
            WhichFastField::Deferred(s, _) => s.clone(),
            WhichFastField::DeferredCtid(alias) => alias.clone(),
            WhichFastField::MatchTag(alias) => alias.clone(),
        }
    }

    /// Returns the SearchFieldType if this is a Named fast field, None otherwise.
    pub fn field_type(&self) -> Option<&SearchFieldType> {
        match self {
            WhichFastField::Named(_, field_type) => Some(field_type),
            WhichFastField::Deferred(_, field_type) => Some(field_type),
            WhichFastField::DeferredCtid(_) | WhichFastField::MatchTag(_) => None,
            _ => None,
        }
    }

    /// Returns the Arrow DataType for this fast field.
    pub fn arrow_data_type(&self) -> arrow_schema::DataType {
        use arrow_schema::DataType;
        match self {
            WhichFastField::Ctid => DataType::UInt64,
            WhichFastField::TableOid => DataType::UInt32,
            WhichFastField::Score => DataType::Float32,
            WhichFastField::Named(_, field_type) => field_type.arrow_data_type(),
            WhichFastField::Junk(_) => DataType::Null,
            WhichFastField::Deferred(_, _field_type) => {
                crate::scan::deferred_encode::deferred_union_data_type()
            }
            WhichFastField::DeferredCtid(_) => DataType::UInt64,
            WhichFastField::MatchTag(_) => DataType::Boolean,
        }
    }
}

/// Build an Arrow schema from a list of fast fields.
///
/// This is used by Scanner and ColumnarExecState to create consistent
/// Arrow schemas for DataFusion execution.
pub fn build_arrow_schema(which_fast_fields: &[WhichFastField]) -> arrow_schema::SchemaRef {
    use arrow_schema::{Field, Schema};
    use std::sync::Arc;

    let fields: Vec<Field> = which_fast_fields
        .iter()
        .map(|wff| Field::new(wff.name(), wff.arrow_data_type(), true))
        .collect();
    Arc::new(Schema::new(fields))
}

/// Partitions packed doc addresses by segment ordinal and invokes `process`
/// once per segment (in sorted segment order) with the segment ordinal and
/// its `(row_index, doc_id)` pairs.
///
/// The `packed_iter` argument yields `(row_index, packed_doc_address)`
pub fn for_each_segment<F>(
    num_segments: usize,
    packed_iter: impl Iterator<Item = (usize, u64)>,
    mut process: F,
) -> Result<()>
where
    F: FnMut(SegmentOrdinal, Vec<(usize, DocId)>) -> Result<()>,
{
    let mut by_seg: Vec<Vec<(usize, DocId)>> = vec![Vec::new(); num_segments];
    for (row_idx, packed) in packed_iter {
        let (seg_ord, doc_id) = unpack_doc_address(packed);
        by_seg[seg_ord as usize].push((row_idx, doc_id));
    }
    for (seg_ord, rows) in by_seg.into_iter().enumerate() {
        if rows.is_empty() {
            continue;
        }

        process(seg_ord as SegmentOrdinal, rows)?;
    }
    Ok(())
}

/// Resolve the `ctid` for a single `doc_address` using a cached per-segment [`FFType`].
///
/// On the first call for a given segment, this opens the `ctid` fast-field column and stores
/// it in `cache`.  Subsequent calls for the same segment reuse the cached reader, avoiding the
/// overhead of re-opening the column on every row.
///
/// # Panics
/// Panics if the `ctid` fast field is absent for the given doc (should never happen in a
/// well-formed ParadeDB index).
#[inline]
pub fn resolve_ctid(
    cache: &mut Option<(tantivy::SegmentOrdinal, FFType)>,
    searcher: &tantivy::Searcher,
    doc_address: tantivy::DocAddress,
) -> u64 {
    let seg_ord = doc_address.segment_ord;
    if cache.as_ref().is_none_or(|(o, _)| *o != seg_ord) {
        *cache = Some((
            seg_ord,
            FFType::new_ctid(searcher.segment_reader(seg_ord).fast_fields()),
        ));
    }
    cache
        .as_ref()
        .unwrap()
        .1
        .as_u64(doc_address.doc_id)
        .expect("ctid should be present")
}

pub(crate) const NULL_TERM_ORDINAL: TermOrdinal = u64::MAX;

/// `NULL_TERM_ORDINAL` represents NULL, and will be emitted last in the sorted order.
pub(crate) fn ords_to_string_array(str_ff: StrColumn, term_ords: &UInt64Array) -> Result<ArrayRef> {
    // Enumerate the term ordinals to preserve their positions, and then sort them by ordinal.
    let mut term_ords = term_ords
        .iter()
        .enumerate()
        .map(|(i, maybe_ord)| (i, maybe_ord.unwrap_or(NULL_TERM_ORDINAL)))
        .collect::<Vec<_>>();
    term_ords.sort_unstable_by_key(|(_, term_ord)| *term_ord);

    // Iterate over the sorted term ordinals: as we visit each term ordinal, we will append the
    // term to a StringViewBuilder's data buffer, and record a view to be appended later in sorted
    // order.
    let mut builder = StringViewBuilder::with_capacity(term_ords.len());
    let mut views: Vec<Option<(u32, u32)>> = Vec::with_capacity(term_ords.len());
    views.resize(term_ords.len(), None);

    let mut buffer = Vec::new();
    let mut bytes = Vec::new();
    let mut current_block_addr = str_ff.dictionary().sstable_index.get_block_with_ord(0);
    let mut current_sstable_delta_reader = str_ff
        .dictionary()
        .sstable_delta_reader_block(current_block_addr.clone())
        .expect("Failed to open term dictionary.");
    let mut current_ordinal = 0;
    let mut previous_term: Option<(TermOrdinal, (u32, u32))> = None;
    for (row_idx, ord) in term_ords {
        if ord == NULL_TERM_ORDINAL {
            // NULL_TERM_ORDINAL sorts highest, so all remaining ords will have `None` views, and
            // be appended to the builder as null.
            break;
        }

        // only advance forward if the new ord is different than the one we just processed
        //
        // this allows the input TermOrdinal iterator to contain and reuse duplicates, so long as
        // it's still sorted
        match &previous_term {
            Some((previous_ord, previous_view)) if *previous_ord == ord => {
                // This is the same term ordinal: reuse the previous view.
                views[row_idx] = Some(*previous_view);
                continue;
            }
            // Fall through.
            _ => {}
        }

        // This is a new term ordinal: decode it and append it to the builder.
        assert!(ord >= current_ordinal);
        // check if block changed for new term_ord
        let new_block_addr = str_ff.dictionary().sstable_index.get_block_with_ord(ord);
        if new_block_addr != current_block_addr {
            current_block_addr = new_block_addr;
            current_ordinal = current_block_addr.first_ordinal;
            current_sstable_delta_reader = str_ff
                .dictionary()
                .sstable_delta_reader_block(current_block_addr.clone())
                .map_err(|e| {
                    DataFusionError::Execution(format!(
                        "Failed to fetch next dictionary block: {e}"
                    ))
                })?;
            bytes.clear();
        }

        // Move to ord inside that block
        for _ in current_ordinal..=ord {
            match current_sstable_delta_reader.advance() {
                Ok(true) => {}
                Ok(false) => {
                    return Err(DataFusionError::Execution(format!(
                        "Term ordinal {ord} did not exist in the dictionary."
                    )));
                }
                Err(e) => {
                    return Err(DataFusionError::Execution(format!(
                        "Failed to decode dictionary block: {e}"
                    )));
                }
            }
            bytes.truncate(current_sstable_delta_reader.common_prefix_len());
            bytes.extend_from_slice(current_sstable_delta_reader.suffix());
        }
        current_ordinal = ord + 1;

        // Set the view for this row_idx.
        let offset: u32 = buffer
            .len()
            .try_into()
            .expect("Too many terms requested in `ords_to_string_array`");
        let len: u32 = bytes
            .len()
            .try_into()
            .expect("Single term is too long in `ords_to_string_array`");
        buffer.extend_from_slice(&bytes);
        previous_term = Some((ord, (offset, len)));
        views[row_idx] = Some((offset, len));
    }

    // Append all the rows' views to the builder.
    let block_no = builder.append_block(Buffer::from(buffer));
    for view in views {
        // Each view is an offset and len in our single block, or None for a null.
        match view {
            Some((offset, len)) => unsafe {
                builder.append_view_unchecked(block_no, offset, len);
            },
            None => builder.append_null(),
        }
    }

    Ok(Arc::new(builder.finish()) as ArrayRef)
}

/// Given an unordered collection of TermOrdinals for the given BytesColumn, return a
/// `BinaryViewArray` with one row per input term ordinal (in the input order).
///
/// This is identical to `ords_to_string_array` but uses `BinaryViewBuilder` for binary data.
///
/// `NULL_TERM_ORDINAL` represents NULL, and will be emitted last in the sorted order.
pub(crate) fn ords_to_bytes_array(
    bytes_ff: BytesColumn,
    term_ords: &UInt64Array,
) -> Result<ArrayRef> {
    // Enumerate the term ordinals to preserve their positions, and then sort them by ordinal.
    let mut term_ords = term_ords
        .iter()
        .enumerate()
        .map(|(i, maybe_ord)| (i, maybe_ord.unwrap_or(NULL_TERM_ORDINAL)))
        .collect::<Vec<_>>();
    term_ords.sort_unstable_by_key(|(_, term_ord)| *term_ord);

    // Iterate over the sorted term ordinals: as we visit each term ordinal, we will append the
    // term to a BinaryViewBuilder's data buffer, and record a view to be appended later in sorted
    // order.
    let mut builder = BinaryViewBuilder::with_capacity(term_ords.len());
    let mut views: Vec<Option<(u32, u32)>> = Vec::with_capacity(term_ords.len());
    views.resize(term_ords.len(), None);

    let mut buffer = Vec::new();
    let mut bytes = Vec::new();
    let mut current_block_addr = bytes_ff.dictionary().sstable_index.get_block_with_ord(0);
    let mut current_sstable_delta_reader = bytes_ff
        .dictionary()
        .sstable_delta_reader_block(current_block_addr.clone())
        .expect("Failed to open term dictionary.");
    let mut current_ordinal = 0;
    let mut previous_term: Option<(TermOrdinal, (u32, u32))> = None;
    for (row_idx, ord) in term_ords {
        if ord == NULL_TERM_ORDINAL {
            // NULL_TERM_ORDINAL sorts highest, so all remaining ords will have `None` views, and
            // be appended to the builder as null.
            break;
        }

        // only advance forward if the new ord is different than the one we just processed
        //
        // this allows the input TermOrdinal iterator to contain and reuse duplicates, so long as
        // it's still sorted
        match &previous_term {
            Some((previous_ord, previous_view)) if *previous_ord == ord => {
                // This is the same term ordinal: reuse the previous view.
                views[row_idx] = Some(*previous_view);
                continue;
            }
            // Fall through.
            _ => {}
        }

        // This is a new term ordinal: decode it and append it to the builder.
        assert!(ord >= current_ordinal);
        // check if block changed for new term_ord
        let new_block_addr = bytes_ff.dictionary().sstable_index.get_block_with_ord(ord);
        if new_block_addr != current_block_addr {
            current_block_addr = new_block_addr;
            current_ordinal = current_block_addr.first_ordinal;
            current_sstable_delta_reader = bytes_ff
                .dictionary()
                .sstable_delta_reader_block(current_block_addr.clone())
                .map_err(|e| {
                    DataFusionError::Execution(format!(
                        "Failed to fetch next dictionary block: {e}"
                    ))
                })?;
            bytes.clear();
        }

        // Move to ord inside that block
        for _ in current_ordinal..=ord {
            match current_sstable_delta_reader.advance() {
                Ok(true) => {}
                Ok(false) => {
                    return Err(DataFusionError::Execution(format!(
                        "Term ordinal {ord} did not exist in the dictionary."
                    )));
                }
                Err(e) => {
                    return Err(DataFusionError::Execution(format!(
                        "Failed to decode dictionary block: {e}"
                    )));
                }
            }
            bytes.truncate(current_sstable_delta_reader.common_prefix_len());
            bytes.extend_from_slice(current_sstable_delta_reader.suffix());
        }
        current_ordinal = ord + 1;

        // Set the view for this row_idx.
        let offset: u32 = buffer
            .len()
            .try_into()
            .expect("Too many terms requested in `ords_to_bytes_array`");
        let len: u32 = bytes
            .len()
            .try_into()
            .expect("Single term is too long in `ords_to_bytes_array`");
        buffer.extend_from_slice(&bytes);
        previous_term = Some((ord, (offset, len)));
        views[row_idx] = Some((offset, len));
    }

    // Append all the rows' views to the builder.
    let block_no = builder.append_block(Buffer::from(buffer));
    for view in views {
        // Each view is an offset and len in our single block, or None for a null.
        match view {
            Some((offset, len)) => unsafe {
                builder.append_view_unchecked(block_no, offset, len);
            },
            None => builder.append_null(),
        }
    }

    Ok(Arc::new(builder.finish()) as ArrayRef)
}

#[cfg(any(test, feature = "pg_test"))]
#[pgrx::pg_schema]
mod tests {
    use super::*;
    use crate::index::mvcc::MvccSatisfies;
    use pgrx::prelude::*;

    /// The helper opens a segment's fast fields only when a value is actually read from that
    /// segment, and stays valid after its source reader is dropped. The fixture includes a
    /// mutable segment, where an open also materializes the segment from the heap; it must
    /// stay cold when only an immutable segment is read.
    #[pg_test]
    fn ffhelper_opens_only_the_segment_it_reads() {
        let (index_rel, _heap) = crate::index::reader::index::test_support::segmented_index_fixture(
            "ffhelper_lazy_test",
            2,
            /* with_mutable */ true,
        );
        let reader = SearchIndexReader::empty(&index_rel, MvccSatisfies::Snapshot).unwrap();
        assert_eq!(
            reader.segment_readers().len(),
            3,
            "two frozen batches plus one mutable segment"
        );
        // Segment ordering is not contractual: pick an immutable segment by its size (the
        // frozen batches hold 10 docs, the mutable segment 5).
        let immutable_ord = reader
            .segment_readers()
            .iter()
            .position(|r| r.num_docs() == 10)
            .expect("an immutable batch segment") as SegmentOrdinal;

        let helper = FFHelper::with_fields(
            &reader,
            &[WhichFastField::Named(
                "id".to_string(),
                SearchFieldType::I64(pg_sys::INT8OID),
            )],
        );
        assert!(helper
            .caches()
            .iter()
            .all(|cache| cache.columns[0].get().is_none() && cache.ctid.get().is_none()));

        // Worker setup drops its source reader before the helper is used.
        drop(reader);
        let value = helper.value(0, DocAddress::new(immutable_ord, 0)).unwrap();
        assert!(i64::try_from(value).is_ok());

        let initialized: Vec<usize> = helper
            .caches()
            .iter()
            .enumerate()
            .filter(|(_, cache)| cache.columns[0].get().is_some() || cache.ctid.get().is_some())
            .map(|(ord, _)| ord)
            .collect();
        assert_eq!(
            initialized,
            vec![immutable_ord as usize],
            "only the segment actually read may open; the mutable segment must stay cold"
        );

        let first = helper.column(immutable_ord, 0) as *const FFType;
        let second = helper.column(immutable_ord, 0) as *const FFType;
        assert_eq!(first, second);
    }
}
