polaranges 0.3.2

use std::collections::HashSet;

use polars_core::prelude::{AnyValue, Column, DataFrame, DataType, NamedFrom, Series};
use ruranges_core::{
    boundary, cluster, complement_single, extend, group_cumsum, max_disjoint, merge, nearest,
    outside_bounds, overlaps, sorts, split, subtract, tile,
};
use rustc_hash::FxHashMap;

use crate::error::{RangeFrameError, Result};
use crate::factorize::{factorize, factorize_pair, factorize_pair_by};
use crate::range_frame::{extract_coordinates, take_rows, RangeView};

#[derive(Clone, Debug)]
pub struct MergeOptions {
    pub count_column: Option<String>,
    pub match_by: Vec<String>,
    pub slack: i64,
}

impl Default for MergeOptions {
    fn default() -> Self {
        Self {
            count_column: None,
            match_by: Vec::new(),
            slack: 0,
        }
    }
}

impl MergeOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct ClusterOptions {
    pub match_by: Vec<String>,
    pub cluster_column: String,
    pub slack: i64,
}

impl Default for ClusterOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            cluster_column: "Cluster".to_owned(),
            slack: 0,
        }
    }
}

impl ClusterOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct CountOverlapsOptions {
    pub match_by: Vec<String>,
    pub slack: i64,
}

impl Default for CountOverlapsOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            slack: 0,
        }
    }
}

impl CountOverlapsOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum JoinType {
    Inner,
    Left,
    Right,
    Outer,
}

#[derive(Clone, Debug)]
pub struct JoinOverlapsOptions {
    pub match_by: Vec<String>,
    pub multiple: crate::OverlapMode,
    pub slack: i64,
    pub suffix: String,
    pub contained_intervals_only: bool,
    pub join_type: JoinType,
    pub report_overlap_column: Option<String>,
    pub preserve_input_order: bool,
}

impl Default for JoinOverlapsOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            multiple: crate::OverlapMode::All,
            slack: 0,
            suffix: "_b".to_owned(),
            contained_intervals_only: false,
            join_type: JoinType::Inner,
            report_overlap_column: None,
            preserve_input_order: true,
        }
    }
}

impl JoinOverlapsOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct IntersectOverlapsOptions {
    pub match_by: Vec<String>,
    pub multiple: crate::OverlapMode,
    pub slack: i64,
    pub contained_intervals_only: bool,
    pub preserve_input_order: bool,
}

impl Default for IntersectOverlapsOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            multiple: crate::OverlapMode::All,
            slack: 0,
            contained_intervals_only: false,
            preserve_input_order: true,
        }
    }
}

impl IntersectOverlapsOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct SetIntersectOverlapsOptions {
    pub match_by: Vec<String>,
    pub multiple: crate::OverlapMode,
    pub preserve_input_order: bool,
}

impl Default for SetIntersectOverlapsOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            multiple: crate::OverlapMode::All,
            preserve_input_order: true,
        }
    }
}

impl SetIntersectOverlapsOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug, Default)]
pub struct SetUnionOverlapsOptions {
    pub match_by: Vec<String>,
}

impl SetUnionOverlapsOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum NearestDirection {
    Any,
    Forward,
    Backward,
}

impl NearestDirection {
    fn as_kernel_str(self) -> &'static str {
        match self {
            Self::Any => "any",
            Self::Forward => "forward",
            Self::Backward => "backward",
        }
    }
}

#[derive(Clone, Debug)]
pub struct NearestOptions {
    pub direction: NearestDirection,
    pub k: usize,
    pub exclude_overlaps: bool,
    pub preserve_input_order: bool,
    pub suffix: String,
    pub distance_column: Option<String>,
    pub match_by: Vec<String>,
    pub left_start_col: String,
    pub left_end_col: String,
    pub right_start_col: String,
    pub right_end_col: String,
}

impl Default for NearestOptions {
    fn default() -> Self {
        Self {
            direction: NearestDirection::Any,
            k: 1,
            exclude_overlaps: false,
            preserve_input_order: true,
            suffix: "_b".to_owned(),
            distance_column: Some("Distance".to_owned()),
            match_by: Vec::new(),
            left_start_col: "Start".to_owned(),
            left_end_col: "End".to_owned(),
            right_start_col: "Start".to_owned(),
            right_end_col: "End".to_owned(),
        }
    }
}

impl NearestOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }

    pub fn with_interval_columns(
        mut self,
        start_col: impl Into<String>,
        end_col: impl Into<String>,
    ) -> Self {
        let start = start_col.into();
        let end = end_col.into();
        self.left_start_col = start.clone();
        self.left_end_col = end.clone();
        self.right_start_col = start;
        self.right_end_col = end;
        self
    }

    pub fn with_left_interval_columns(
        mut self,
        start_col: impl Into<String>,
        end_col: impl Into<String>,
    ) -> Self {
        self.left_start_col = start_col.into();
        self.left_end_col = end_col.into();
        self
    }

    pub fn with_right_interval_columns(
        mut self,
        start_col: impl Into<String>,
        end_col: impl Into<String>,
    ) -> Self {
        self.right_start_col = start_col.into();
        self.right_end_col = end_col.into();
        self
    }
}

#[derive(Clone, Debug)]
pub struct SubtractOptions {
    pub match_by: Vec<String>,
    pub preserve_input_order: bool,
}

#[derive(Clone, Debug)]
pub struct SortRangesOptions {
    pub match_by: Vec<String>,
    pub negative_strand_column: Option<String>,
}

impl Default for SortRangesOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            negative_strand_column: None,
        }
    }
}

impl SortRangesOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug, Default)]
pub struct ExtendRangesOptions {
    pub match_by: Vec<String>,
    pub ext: Option<i64>,
    pub ext_3: Option<i64>,
    pub ext_5: Option<i64>,
    pub negative_strand_column: Option<String>,
}

impl ExtendRangesOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct TileRangesOptions {
    pub match_by: Vec<String>,
    pub tile_size: i64,
    pub negative_strand_column: Option<String>,
    pub overlap_column: Option<String>,
}

impl TileRangesOptions {
    pub fn new(tile_size: i64) -> Self {
        Self {
            match_by: Vec::new(),
            tile_size,
            negative_strand_column: None,
            overlap_column: None,
        }
    }

    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct ClipRangesOptions {
    pub chromsizes: Option<FxHashMap<String, i64>>,
    pub remove: bool,
    pub only_right: bool,
    pub group_sizes_col: String,
}

impl Default for ClipRangesOptions {
    fn default() -> Self {
        Self {
            chromsizes: None,
            remove: false,
            only_right: false,
            group_sizes_col: "Chromosome".to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct GroupCumsumOptions {
    pub match_by: Vec<String>,
    pub forward_strand_column: Option<String>,
    pub cumsum_start_column: Option<String>,
    pub cumsum_end_column: Option<String>,
    pub keep_order: bool,
}

impl Default for GroupCumsumOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            forward_strand_column: None,
            cumsum_start_column: None,
            cumsum_end_column: None,
            keep_order: true,
        }
    }
}

impl GroupCumsumOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct MaxDisjointOptions {
    pub match_by: Vec<String>,
    pub slack: i64,
    pub preserve_input_order: bool,
}

impl Default for MaxDisjointOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            slack: 0,
            preserve_input_order: true,
        }
    }
}

impl MaxDisjointOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct SplitOverlapsOptions {
    pub match_by: Vec<String>,
    pub between: bool,
}

impl Default for SplitOverlapsOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            between: false,
        }
    }
}

impl SplitOverlapsOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug, Default)]
pub struct OuterRangesOptions {
    pub match_by: Vec<String>,
}

impl OuterRangesOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

#[derive(Clone, Debug)]
pub struct ComplementRangesOptions {
    pub match_by: Vec<String>,
    pub include_first_interval: bool,
    pub chromsizes: Option<FxHashMap<String, i64>>,
    pub group_sizes_col: String,
}

impl Default for ComplementRangesOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            include_first_interval: false,
            chromsizes: None,
            group_sizes_col: "Chromosome".to_owned(),
        }
    }
}

impl ComplementRangesOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

impl Default for SubtractOptions {
    fn default() -> Self {
        Self {
            match_by: Vec::new(),
            preserve_input_order: true,
        }
    }
}

impl SubtractOptions {
    pub fn with_match_by<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.match_by = columns.into_iter().map(Into::into).collect();
        self
    }
}

impl<'a> RangeView<'a> {
    pub(crate) fn merge_overlaps(&self, options: &MergeOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            if let Ok(slack) = i32::try_from(options.slack) {
                let (idx, merged_starts, merged_ends, counts) =
                    merge::sweep_line_merge(&groups, starts, ends, slack);
                return build_merge_result_i32(
                    self,
                    &idx,
                    merged_starts,
                    merged_ends,
                    counts,
                    options,
                );
            }
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let (idx, merged_starts, merged_ends, counts) =
                merge::sweep_line_merge(&groups, starts, ends, options.slack);
            return build_merge_result_i64(self, &idx, merged_starts, merged_ends, counts, options);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let (idx, merged_starts, merged_ends, counts) =
            merge::sweep_line_merge(&groups, &starts, &ends, options.slack);
        build_merge_result_i64(self, &idx, merged_starts, merged_ends, counts, options)
    }

    pub(crate) fn cluster_overlaps(&self, options: &ClusterOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            if let Ok(slack) = i32::try_from(options.slack) {
                let (cluster_ids, idx) = cluster::sweep_line_cluster(&groups, starts, ends, slack);
                return build_cluster_result(self.df, &idx, cluster_ids, &options.cluster_column);
            }
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let (cluster_ids, idx) =
                cluster::sweep_line_cluster(&groups, starts, ends, options.slack);
            return build_cluster_result(self.df, &idx, cluster_ids, &options.cluster_column);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let (cluster_ids, idx) =
            cluster::sweep_line_cluster(&groups, &starts, &ends, options.slack);
        build_cluster_result(self.df, &idx, cluster_ids, &options.cluster_column)
    }

    pub(crate) fn count_overlaps(
        &self,
        other: &RangeView<'_>,
        options: &CountOverlapsOptions,
    ) -> Result<Series> {
        let (left_groups, right_groups) = factorize_pair(self.df, other.df, &options.match_by)?;
        let left_coords = extract_coordinates(self.df, self.start_col, self.end_col)?;
        let right_coords = extract_coordinates(other.df, other.start_col, other.end_col)?;

        if let (Some((left_starts, left_ends)), Some((right_starts, right_ends))) =
            (left_coords.as_i32(), right_coords.as_i32())
        {
            if let Ok(slack) = i32::try_from(options.slack) {
                let counts = overlaps::count_overlaps(
                    &left_groups,
                    left_starts,
                    left_ends,
                    &right_groups,
                    right_starts,
                    right_ends,
                    slack,
                );
                return Ok(Series::new("Count".into(), counts));
            }
        }

        if let (Some((left_starts, left_ends)), Some((right_starts, right_ends))) =
            (left_coords.as_i64(), right_coords.as_i64())
        {
            let counts = overlaps::count_overlaps(
                &left_groups,
                left_starts,
                left_ends,
                &right_groups,
                right_starts,
                right_ends,
                options.slack,
            );
            return Ok(Series::new("Count".into(), counts));
        }

        let (left_starts, left_ends) = left_coords.into_i64_buffers();
        let (right_starts, right_ends) = right_coords.into_i64_buffers();
        let counts = overlaps::count_overlaps(
            &left_groups,
            &left_starts,
            &left_ends,
            &right_groups,
            &right_starts,
            &right_ends,
            options.slack,
        );
        Ok(Series::new("Count".into(), counts))
    }

    pub(crate) fn join_overlaps(
        &self,
        other: &RangeView<'_>,
        options: &JoinOverlapsOptions,
    ) -> Result<DataFrame> {
        let overlap_options = crate::OverlapOptions {
            multiple: options.multiple,
            slack: options.slack,
            contained_intervals_only: options.contained_intervals_only,
            match_by: options.match_by.clone(),
            preserve_input_order: options.preserve_input_order,
            ..crate::OverlapOptions::default()
        };
        let pairs = self.overlap_pairs(other, &overlap_options)?;
        build_join_result(self.df, other.df, &pairs.left, &pairs.right, options)
    }

    pub(crate) fn intersect_overlaps(
        &self,
        other: &RangeView<'_>,
        options: &IntersectOverlapsOptions,
    ) -> Result<DataFrame> {
        let overlap_options = crate::OverlapOptions {
            multiple: options.multiple,
            slack: options.slack,
            contained_intervals_only: options.contained_intervals_only,
            match_by: options.match_by.clone(),
            preserve_input_order: options.preserve_input_order,
            ..crate::OverlapOptions::default()
        };
        let pairs = self.overlap_pairs(other, &overlap_options)?;
        build_intersect_result(self, other, &pairs.left, &pairs.right)
    }

    pub(crate) fn set_intersect_overlaps(
        &self,
        other: &RangeView<'_>,
        options: &SetIntersectOverlapsOptions,
    ) -> Result<DataFrame> {
        let left_merged = self.merge_overlaps(&MergeOptions {
            count_column: None,
            match_by: options.match_by.clone(),
            slack: 0,
        })?;
        let right_merged = other.merge_overlaps(&MergeOptions {
            count_column: None,
            match_by: options.match_by.clone(),
            slack: 0,
        })?;
        let left = RangeView::with_columns(&left_merged, self.start_col, self.end_col)?;
        let right = RangeView::with_columns(&right_merged, other.start_col, other.end_col)?;
        left.intersect_overlaps(
            &right,
            &IntersectOverlapsOptions {
                match_by: options.match_by.clone(),
                multiple: options.multiple,
                slack: 0,
                contained_intervals_only: false,
                preserve_input_order: options.preserve_input_order,
            },
        )
    }

    pub(crate) fn set_union_overlaps(
        &self,
        other: &RangeView<'_>,
        options: &SetUnionOverlapsOptions,
    ) -> Result<DataFrame> {
        let _ = factorize_pair(self.df, other.df, &options.match_by)?;
        let left_idx = all_indices(self.df.height())?;
        let right_idx = all_indices(other.df.height())?;
        let mut combined = take_merge_projection(
            self.df,
            &left_idx,
            self.start_col,
            self.end_col,
            &options.match_by,
        )?;
        let right = take_merge_projection(
            other.df,
            &right_idx,
            other.start_col,
            other.end_col,
            &options.match_by,
        )?;
        combined.vstack_mut(&right)?;
        let combined = RangeView::with_columns(&combined, self.start_col, self.end_col)?;
        combined.merge_overlaps(&MergeOptions {
            count_column: None,
            match_by: options.match_by.clone(),
            slack: 0,
        })
    }

    pub(crate) fn sort_ranges(&self, options: &SortRangesOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let reverse = match options.negative_strand_column.as_deref() {
            Some(column) => Some(negative_strand_flags(self.df, column)?),
            None => None,
        };
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            let idx = sorts::sort_order_idx(&groups, starts, ends, reverse.as_deref());
            return take_rows(self.df, &idx);
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let idx = sorts::sort_order_idx(&groups, starts, ends, reverse.as_deref());
            return take_rows(self.df, &idx);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let idx = sorts::sort_order_idx(&groups, &starts, &ends, reverse.as_deref());
        take_rows(self.df, &idx)
    }

    pub(crate) fn extend_ranges(&self, options: &ExtendRangesOptions) -> Result<DataFrame> {
        let (ext_3, ext_5) = resolve_extend_options(options)?;
        let groups = if options.match_by.is_empty() {
            all_indices(self.df.height())?
        } else {
            factorize(self.df, &options.match_by)?
        };
        let negative_strand = match options.negative_strand_column.as_deref() {
            Some(column) => negative_strand_flags(self.df, column)?,
            None => vec![false; self.df.height()],
        };
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        let mut out = self.df.clone();
        if let Some((starts, ends)) = coords.as_i32() {
            if let (Ok(ext_3), Ok(ext_5)) = (i32::try_from(ext_3), i32::try_from(ext_5)) {
                let (starts, ends) =
                    extend::extend_grp(&groups, starts, ends, &negative_strand, ext_3, ext_5);
                replace_interval_columns_i32(&mut out, self.start_col, self.end_col, starts, ends)?;
                return Ok(out);
            }
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let (starts, ends) =
                extend::extend_grp(&groups, starts, ends, &negative_strand, ext_3, ext_5);
            replace_interval_columns_i64(&mut out, self.start_col, self.end_col, starts, ends)?;
            return Ok(out);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let (starts, ends) =
            extend::extend_grp(&groups, &starts, &ends, &negative_strand, ext_3, ext_5);
        replace_interval_columns_i64(&mut out, self.start_col, self.end_col, starts, ends)?;
        Ok(out)
    }

    pub(crate) fn tile_ranges(&self, options: &TileRangesOptions) -> Result<DataFrame> {
        if options.tile_size <= 0 {
            return Err(compute_error("tile_size must be greater than 0"));
        }

        let _ = factorize(self.df, &options.match_by)?;
        let negative_strand = match options.negative_strand_column.as_deref() {
            Some(column) => negative_strand_flags(self.df, column)?,
            None => vec![false; self.df.height()],
        };
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            if let Ok(tile_size) = i32::try_from(options.tile_size) {
                let (starts, ends, idx, overlaps) =
                    tile::tile(starts, ends, &negative_strand, tile_size);
                return build_tile_result_i32(self, &idx, starts, ends, overlaps, options);
            }
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let (starts, ends, idx, overlaps) =
                tile::tile(starts, ends, &negative_strand, options.tile_size);
            return build_tile_result_i64(self, &idx, starts, ends, overlaps, options);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let (starts, ends, idx, overlaps) =
            tile::tile(&starts, &ends, &negative_strand, options.tile_size);
        build_tile_result_i64(self, &idx, starts, ends, overlaps, options)
    }

    pub(crate) fn clip_ranges(&self, options: &ClipRangesOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, std::slice::from_ref(&options.group_sizes_col))?;
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            let chrom_lens =
                row_aligned_bounds_i32(self.df, &options.group_sizes_col, ends, options)?;
            let (idx, starts, ends) = outside_bounds::outside_bounds(
                &groups,
                starts,
                ends,
                &chrom_lens,
                !options.remove,
                options.only_right,
            )
            .map_err(compute_error)?;
            return build_clip_result_i32(self, &idx, starts, ends, options);
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let chrom_lens =
                row_aligned_bounds_i64(self.df, &options.group_sizes_col, ends, options)?;
            let (idx, starts, ends) = outside_bounds::outside_bounds(
                &groups,
                starts,
                ends,
                &chrom_lens,
                !options.remove,
                options.only_right,
            )
            .map_err(compute_error)?;
            return build_clip_result_i64(self, &idx, starts, ends, options);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let chrom_lens = row_aligned_bounds_i64(self.df, &options.group_sizes_col, &ends, options)?;
        let (idx, starts, ends) = outside_bounds::outside_bounds(
            &groups,
            &starts,
            &ends,
            &chrom_lens,
            !options.remove,
            options.only_right,
        )
        .map_err(compute_error)?;
        build_clip_result_i64(self, &idx, starts, ends, options)
    }

    pub(crate) fn group_cumsum(&self, options: &GroupCumsumOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let forward_strand = match options.forward_strand_column.as_deref() {
            Some(column) => forward_strand_flags(self.df, column)?,
            None => vec![true; self.df.height()],
        };
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            let (idx, cumsum_starts, cumsum_ends) = group_cumsum::sweep_line_cumsum(
                &groups,
                starts,
                ends,
                &forward_strand,
                options.keep_order,
            );
            return build_group_cumsum_result_i32(self, &idx, cumsum_starts, cumsum_ends, options);
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let (idx, cumsum_starts, cumsum_ends) = group_cumsum::sweep_line_cumsum(
                &groups,
                starts,
                ends,
                &forward_strand,
                options.keep_order,
            );
            return build_group_cumsum_result_i64(self, &idx, cumsum_starts, cumsum_ends, options);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let (idx, cumsum_starts, cumsum_ends) = group_cumsum::sweep_line_cumsum(
            &groups,
            &starts,
            &ends,
            &forward_strand,
            options.keep_order,
        );
        build_group_cumsum_result_i64(self, &idx, cumsum_starts, cumsum_ends, options)
    }

    pub(crate) fn max_disjoint_overlaps(&self, options: &MaxDisjointOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            if let Ok(slack) = i32::try_from(options.slack) {
                let idx = max_disjoint::max_disjoint(
                    &groups,
                    starts,
                    ends,
                    slack,
                    options.preserve_input_order,
                );
                return take_rows(self.df, &idx);
            }
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let idx = max_disjoint::max_disjoint(
                &groups,
                starts,
                ends,
                options.slack,
                options.preserve_input_order,
            );
            return take_rows(self.df, &idx);
        }

        let (starts, ends) = coords.into_i64_buffers();
        let idx = max_disjoint::max_disjoint(
            &groups,
            &starts,
            &ends,
            options.slack,
            options.preserve_input_order,
        );
        take_rows(self.df, &idx)
    }

    pub(crate) fn split_overlaps(&self, options: &SplitOverlapsOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            let (idx, starts, ends) =
                split::sweep_line_split(&groups, starts, ends, 0_i32, options.between);
            return build_projected_interval_result_i32(
                self,
                &idx,
                starts,
                ends,
                &options.match_by,
            );
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let (idx, starts, ends) =
                split::sweep_line_split(&groups, starts, ends, 0_i64, options.between);
            return build_projected_interval_result_i64(
                self,
                &idx,
                starts,
                ends,
                &options.match_by,
            );
        }

        let (starts, ends) = coords.into_i64_buffers();
        let (idx, starts, ends) =
            split::sweep_line_split(&groups, &starts, &ends, 0_i64, options.between);
        build_projected_interval_result_i64(self, &idx, starts, ends, &options.match_by)
    }

    pub(crate) fn outer_ranges(&self, options: &OuterRangesOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            let (idx, starts, ends, _counts) = boundary::sweep_line_boundary(&groups, starts, ends);
            return build_projected_interval_result_i32(
                self,
                &idx,
                starts,
                ends,
                &options.match_by,
            );
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let (idx, starts, ends, _counts) = boundary::sweep_line_boundary(&groups, starts, ends);
            return build_projected_interval_result_i64(
                self,
                &idx,
                starts,
                ends,
                &options.match_by,
            );
        }

        let (starts, ends) = coords.into_i64_buffers();
        let (idx, starts, ends, _counts) = boundary::sweep_line_boundary(&groups, &starts, &ends);
        build_projected_interval_result_i64(self, &idx, starts, ends, &options.match_by)
    }

    pub(crate) fn complement_ranges(&self, options: &ComplementRangesOptions) -> Result<DataFrame> {
        let merged = self.merge_overlaps(&MergeOptions {
            count_column: None,
            match_by: options.match_by.clone(),
            slack: 0,
        })?;
        let view = RangeView::with_columns(&merged, self.start_col, self.end_col)?;
        view.complement_ranges_merged(options)
    }

    fn complement_ranges_merged(&self, options: &ComplementRangesOptions) -> Result<DataFrame> {
        let groups = factorize(self.df, &options.match_by)?;
        let coords = extract_coordinates(self.df, self.start_col, self.end_col)?;

        if let Some((starts, ends)) = coords.as_i32() {
            let chrom_lens = complement_lengths_i32(self.df, &groups, options)?;
            let (_groups, starts, ends, idx) = complement_single::sweep_line_complement(
                &groups,
                starts,
                ends,
                0_i32,
                &chrom_lens,
                options.include_first_interval,
            );
            return build_projected_interval_result_i32(
                self,
                &idx,
                starts,
                ends,
                &options.match_by,
            );
        }

        if let Some((starts, ends)) = coords.as_i64() {
            let chrom_lens = complement_lengths_i64(self.df, &groups, options)?;
            let (_groups, starts, ends, idx) = complement_single::sweep_line_complement(
                &groups,
                starts,
                ends,
                0_i64,
                &chrom_lens,
                options.include_first_interval,
            );
            return build_projected_interval_result_i64(
                self,
                &idx,
                starts,
                ends,
                &options.match_by,
            );
        }

        let (starts, ends) = coords.into_i64_buffers();
        let chrom_lens = complement_lengths_i64(self.df, &groups, options)?;
        let (_groups, starts, ends, idx) = complement_single::sweep_line_complement(
            &groups,
            &starts,
            &ends,
            0_i64,
            &chrom_lens,
            options.include_first_interval,
        );
        build_projected_interval_result_i64(self, &idx, starts, ends, &options.match_by)
    }

    pub(crate) fn nearest_ranges_by_match(
        &self,
        other: &RangeView<'_>,
        left_match_by: &[String],
        right_match_by: &[String],
        options: &NearestOptions,
    ) -> Result<DataFrame> {
        if options.k == 0 {
            return Err(RangeFrameError::InvalidNearestK { k: options.k });
        }

        let (left_groups, right_groups) =
            factorize_pair_by(self.df, other.df, left_match_by, right_match_by)?;
        let left_coords = extract_coordinates(self.df, self.start_col, self.end_col)?;
        let right_coords = extract_coordinates(other.df, other.start_col, other.end_col)?;

        if let (Some((left_starts, left_ends)), Some((right_starts, right_ends))) =
            (left_coords.as_i32(), right_coords.as_i32())
        {
            let (left_idx, right_idx, distances) = nearest::nearest(
                &left_groups,
                left_starts,
                left_ends,
                &right_groups,
                right_starts,
                right_ends,
                0_i32,
                options.k,
                !options.exclude_overlaps,
                options.direction.as_kernel_str(),
                options.preserve_input_order,
            );
            return build_nearest_result_i32(
                self.df, other.df, &left_idx, &right_idx, distances, options,
            );
        }

        if let (Some((left_starts, left_ends)), Some((right_starts, right_ends))) =
            (left_coords.as_i64(), right_coords.as_i64())
        {
            let (left_idx, right_idx, distances) = nearest::nearest(
                &left_groups,
                left_starts,
                left_ends,
                &right_groups,
                right_starts,
                right_ends,
                0_i64,
                options.k,
                !options.exclude_overlaps,
                options.direction.as_kernel_str(),
                options.preserve_input_order,
            );
            return build_nearest_result_i64(
                self.df, other.df, &left_idx, &right_idx, distances, options,
            );
        }

        let (left_starts, left_ends) = left_coords.into_i64_buffers();
        let (right_starts, right_ends) = right_coords.into_i64_buffers();
        let (left_idx, right_idx, distances) = nearest::nearest(
            &left_groups,
            &left_starts,
            &left_ends,
            &right_groups,
            &right_starts,
            &right_ends,
            0_i64,
            options.k,
            !options.exclude_overlaps,
            options.direction.as_kernel_str(),
            options.preserve_input_order,
        );
        build_nearest_result_i64(self.df, other.df, &left_idx, &right_idx, distances, options)
    }

    pub(crate) fn subtract_overlaps(
        &self,
        other: &RangeView<'_>,
        options: &SubtractOptions,
    ) -> Result<DataFrame> {
        let (left_groups, right_groups) = factorize_pair(self.df, other.df, &options.match_by)?;
        let left_coords = extract_coordinates(self.df, self.start_col, self.end_col)?;
        let right_coords = extract_coordinates(other.df, other.start_col, other.end_col)?;

        if let (Some((left_starts, left_ends)), Some((right_starts, right_ends))) =
            (left_coords.as_i32(), right_coords.as_i32())
        {
            let (idx, starts, ends) = subtract::sweep_line_subtract(
                &left_groups,
                left_starts,
                left_ends,
                &right_groups,
                right_starts,
                right_ends,
                options.preserve_input_order,
            );
            return build_subtract_result_i32(self, &idx, starts, ends);
        }

        if let (Some((left_starts, left_ends)), Some((right_starts, right_ends))) =
            (left_coords.as_i64(), right_coords.as_i64())
        {
            let (idx, starts, ends) = subtract::sweep_line_subtract(
                &left_groups,
                left_starts,
                left_ends,
                &right_groups,
                right_starts,
                right_ends,
                options.preserve_input_order,
            );
            return build_subtract_result_i64(self, &idx, starts, ends);
        }

        let (left_starts, left_ends) = left_coords.into_i64_buffers();
        let (right_starts, right_ends) = right_coords.into_i64_buffers();
        let (idx, starts, ends) = subtract::sweep_line_subtract(
            &left_groups,
            &left_starts,
            &left_ends,
            &right_groups,
            &right_starts,
            &right_ends,
            options.preserve_input_order,
        );
        build_subtract_result_i64(self, &idx, starts, ends)
    }
}

fn build_merge_result_i32(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i32>,
    ends: Vec<i32>,
    counts: Vec<u32>,
    options: &MergeOptions,
) -> Result<DataFrame> {
    let mut out = take_merge_projection(
        view.df,
        indices,
        view.start_col,
        view.end_col,
        &options.match_by,
    )?;
    replace_interval_columns_i32(&mut out, view.start_col, view.end_col, starts, ends)?;

    if let Some(count_column) = options.count_column.as_deref() {
        out.with_column(Column::from(Series::new(count_column.into(), counts)))?;
    }

    Ok(out)
}

fn build_merge_result_i64(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i64>,
    ends: Vec<i64>,
    counts: Vec<u32>,
    options: &MergeOptions,
) -> Result<DataFrame> {
    let mut out = take_merge_projection(
        view.df,
        indices,
        view.start_col,
        view.end_col,
        &options.match_by,
    )?;
    replace_interval_columns_i64(&mut out, view.start_col, view.end_col, starts, ends)?;

    if let Some(count_column) = options.count_column.as_deref() {
        out.with_column(Column::from(Series::new(count_column.into(), counts)))?;
    }

    Ok(out)
}

fn build_cluster_result(
    df: &DataFrame,
    indices: &[u32],
    cluster_ids: Vec<u32>,
    cluster_column: &str,
) -> Result<DataFrame> {
    let mut out = take_rows(df, indices)?;
    out.with_column(Column::from(Series::new(
        cluster_column.into(),
        cluster_ids,
    )))?;
    Ok(out)
}

fn build_nearest_result_i32(
    left_df: &DataFrame,
    right_df: &DataFrame,
    left_idx: &[u32],
    right_idx: &[u32],
    distances: Vec<i32>,
    options: &NearestOptions,
) -> Result<DataFrame> {
    let mut out = take_rows(left_df, left_idx)?;
    append_suffixed_columns(&mut out, right_df, right_idx, &options.suffix)?;

    if let Some(distance_column) = options.distance_column.as_deref() {
        out.with_column(Column::from(Series::new(distance_column.into(), distances)))?;
    }

    Ok(out)
}

fn build_nearest_result_i64(
    left_df: &DataFrame,
    right_df: &DataFrame,
    left_idx: &[u32],
    right_idx: &[u32],
    distances: Vec<i64>,
    options: &NearestOptions,
) -> Result<DataFrame> {
    let mut out = take_rows(left_df, left_idx)?;
    append_suffixed_columns(&mut out, right_df, right_idx, &options.suffix)?;

    if let Some(distance_column) = options.distance_column.as_deref() {
        out.with_column(Column::from(Series::new(distance_column.into(), distances)))?;
    }

    Ok(out)
}

fn build_subtract_result_i32(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i32>,
    ends: Vec<i32>,
) -> Result<DataFrame> {
    let mut out = take_rows(view.df, indices)?;
    replace_interval_columns_i32(&mut out, view.start_col, view.end_col, starts, ends)?;
    Ok(out)
}

fn build_subtract_result_i64(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i64>,
    ends: Vec<i64>,
) -> Result<DataFrame> {
    let mut out = take_rows(view.df, indices)?;
    replace_interval_columns_i64(&mut out, view.start_col, view.end_col, starts, ends)?;
    Ok(out)
}

fn build_tile_result_i32(
    view: &RangeView<'_>,
    indices: &[usize],
    starts: Vec<i32>,
    ends: Vec<i32>,
    overlaps: Vec<f64>,
    options: &TileRangesOptions,
) -> Result<DataFrame> {
    let indices = usize_indices_to_u32(indices)?;
    let mut out = take_rows(view.df, &indices)?;
    replace_interval_columns_i32(&mut out, view.start_col, view.end_col, starts, ends)?;
    if let Some(column) = options.overlap_column.as_deref() {
        out.with_column(Column::from(Series::new(column.into(), overlaps)))?;
    }
    Ok(out)
}

fn build_tile_result_i64(
    view: &RangeView<'_>,
    indices: &[usize],
    starts: Vec<i64>,
    ends: Vec<i64>,
    overlaps: Vec<f64>,
    options: &TileRangesOptions,
) -> Result<DataFrame> {
    let indices = usize_indices_to_u32(indices)?;
    let mut out = take_rows(view.df, &indices)?;
    replace_interval_columns_i64(&mut out, view.start_col, view.end_col, starts, ends)?;
    if let Some(column) = options.overlap_column.as_deref() {
        out.with_column(Column::from(Series::new(column.into(), overlaps)))?;
    }
    Ok(out)
}

fn build_clip_result_i32(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i32>,
    ends: Vec<i32>,
    options: &ClipRangesOptions,
) -> Result<DataFrame> {
    let mut out = take_rows(view.df, indices)?;
    if !options.remove {
        replace_interval_columns_i32(&mut out, view.start_col, view.end_col, starts, ends)?;
    }
    Ok(out)
}

fn build_clip_result_i64(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i64>,
    ends: Vec<i64>,
    options: &ClipRangesOptions,
) -> Result<DataFrame> {
    let mut out = take_rows(view.df, indices)?;
    if !options.remove {
        replace_interval_columns_i64(&mut out, view.start_col, view.end_col, starts, ends)?;
    }
    Ok(out)
}

fn build_group_cumsum_result_i32(
    view: &RangeView<'_>,
    indices: &[u32],
    cumsum_starts: Vec<i32>,
    cumsum_ends: Vec<i32>,
    options: &GroupCumsumOptions,
) -> Result<DataFrame> {
    let mut out = take_rows(view.df, indices)?;
    append_or_replace_cumsum_i32(
        &mut out,
        view.start_col,
        view.end_col,
        cumsum_starts,
        cumsum_ends,
        options,
    )?;
    Ok(out)
}

fn build_group_cumsum_result_i64(
    view: &RangeView<'_>,
    indices: &[u32],
    cumsum_starts: Vec<i64>,
    cumsum_ends: Vec<i64>,
    options: &GroupCumsumOptions,
) -> Result<DataFrame> {
    let mut out = take_rows(view.df, indices)?;
    append_or_replace_cumsum_i64(
        &mut out,
        view.start_col,
        view.end_col,
        cumsum_starts,
        cumsum_ends,
        options,
    )?;
    Ok(out)
}

fn build_join_result(
    left_df: &DataFrame,
    right_df: &DataFrame,
    left_idx: &[u32],
    right_idx: &[u32],
    options: &JoinOverlapsOptions,
) -> Result<DataFrame> {
    let right_names = renamed_right_column_names(left_df, right_df, &options.suffix);
    let mut frames = Vec::new();

    let mut inner = take_rows(left_df, left_idx)?;
    append_renamed_columns(&mut inner, right_df, right_idx, &right_names)?;
    if let Some(column) = options.report_overlap_column.as_deref() {
        append_overlap_length_column(&mut inner, column, "Start", "End", &options.suffix)?;
    }
    frames.push(inner);

    if matches!(options.join_type, JoinType::Left | JoinType::Outer) {
        let missing = missing_indices(left_df.height(), left_idx);
        if !missing.is_empty() {
            let mut left_missing = take_rows(left_df, &missing)?;
            let nulls = null_columns_for(right_df, &right_names, missing.len());
            left_missing.hstack_mut(&nulls)?;
            append_null_overlap_length_column(&mut left_missing, options, missing.len())?;
            frames.push(left_missing);
        }
    }

    if matches!(options.join_type, JoinType::Right | JoinType::Outer) {
        let missing = missing_indices(right_df.height(), right_idx);
        if !missing.is_empty() {
            let mut right_missing = null_frame_for(left_df, missing.len())?;
            append_renamed_columns(&mut right_missing, right_df, &missing, &right_names)?;
            append_null_overlap_length_column(&mut right_missing, options, missing.len())?;
            frames.push(right_missing);
        }
    }

    concat_same_schema(frames)
}

fn build_intersect_result(
    left: &RangeView<'_>,
    right: &RangeView<'_>,
    left_idx: &[u32],
    right_idx: &[u32],
) -> Result<DataFrame> {
    let mut out = take_rows(left.df, left_idx)?;
    let left_taken = take_rows(left.df, left_idx)?;
    let right_taken = take_rows(right.df, right_idx)?;
    let left_coords = extract_coordinates(&left_taken, left.start_col, left.end_col)?;
    let right_coords = extract_coordinates(&right_taken, right.start_col, right.end_col)?;
    let (left_starts, left_ends) = left_coords.into_i64_buffers();
    let (right_starts, right_ends) = right_coords.into_i64_buffers();

    let starts = left_starts
        .into_iter()
        .zip(right_starts)
        .map(|(left, right)| left.max(right))
        .collect::<Vec<_>>();
    let ends = left_ends
        .into_iter()
        .zip(right_ends)
        .map(|(left, right)| left.min(right))
        .collect::<Vec<_>>();

    replace_interval_columns_i64(&mut out, left.start_col, left.end_col, starts, ends)?;
    Ok(out)
}

fn build_projected_interval_result_i32(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i32>,
    ends: Vec<i32>,
    match_by: &[String],
) -> Result<DataFrame> {
    let mut out =
        take_interval_projection(view.df, indices, view.start_col, view.end_col, match_by)?;
    replace_interval_columns_i32(&mut out, view.start_col, view.end_col, starts, ends)?;
    Ok(out)
}

fn build_projected_interval_result_i64(
    view: &RangeView<'_>,
    indices: &[u32],
    starts: Vec<i64>,
    ends: Vec<i64>,
    match_by: &[String],
) -> Result<DataFrame> {
    let mut out =
        take_interval_projection(view.df, indices, view.start_col, view.end_col, match_by)?;
    replace_interval_columns_i64(&mut out, view.start_col, view.end_col, starts, ends)?;
    Ok(out)
}

fn take_merge_projection(
    df: &DataFrame,
    indices: &[u32],
    start_col: &str,
    end_col: &str,
    match_by: &[String],
) -> Result<DataFrame> {
    let taken = take_rows(df, indices)?;
    let columns = taken
        .columns()
        .iter()
        .filter(|column| {
            let name = column.name().as_str();
            name == start_col || name == end_col || match_by.iter().any(|value| value == name)
        })
        .cloned()
        .collect::<Vec<_>>();
    Ok(DataFrame::new_infer_height(columns)?)
}

fn take_interval_projection(
    df: &DataFrame,
    indices: &[u32],
    start_col: &str,
    end_col: &str,
    match_by: &[String],
) -> Result<DataFrame> {
    take_merge_projection(df, indices, start_col, end_col, match_by)
}

fn append_suffixed_columns(
    left: &mut DataFrame,
    right_df: &DataFrame,
    right_idx: &[u32],
    suffix: &str,
) -> Result<()> {
    let right_taken = take_rows(right_df, right_idx)?;
    let mut columns = Vec::with_capacity(right_taken.width());

    for column in right_taken.columns() {
        let mut column = column.clone();
        column.rename(format!("{}{}", column.name(), suffix).into());
        columns.push(column);
    }

    left.hstack_mut(&columns)?;
    Ok(())
}

fn renamed_right_column_names(left: &DataFrame, right: &DataFrame, suffix: &str) -> Vec<String> {
    let left_names = left
        .get_column_names()
        .into_iter()
        .map(|name| name.as_str().to_owned())
        .collect::<HashSet<_>>();

    right
        .get_column_names()
        .into_iter()
        .map(|name| {
            if left_names.contains(name.as_str()) {
                format!("{}{}", name.as_str(), suffix)
            } else {
                name.as_str().to_owned()
            }
        })
        .collect()
}

fn append_renamed_columns(
    left: &mut DataFrame,
    right_df: &DataFrame,
    right_idx: &[u32],
    names: &[String],
) -> Result<()> {
    let right_taken = take_rows(right_df, right_idx)?;
    let mut columns = Vec::with_capacity(right_taken.width());

    for (column, name) in right_taken.columns().iter().zip(names) {
        let mut column = column.clone();
        column.rename(name.as_str().into());
        columns.push(column);
    }

    left.hstack_mut(&columns)?;
    Ok(())
}

fn null_columns_for(df: &DataFrame, names: &[String], len: usize) -> Vec<Column> {
    df.columns()
        .iter()
        .zip(names)
        .map(|(column, name)| Column::full_null(name.as_str().into(), len, column.dtype()))
        .collect()
}

fn null_frame_for(df: &DataFrame, len: usize) -> Result<DataFrame> {
    let columns = df
        .columns()
        .iter()
        .map(|column| Column::full_null(column.name().clone(), len, column.dtype()))
        .collect::<Vec<_>>();
    Ok(DataFrame::new_infer_height(columns)?)
}

fn missing_indices(len: usize, present: &[u32]) -> Vec<u32> {
    let mut seen = vec![false; len];
    for &idx in present {
        if let Some(slot) = seen.get_mut(idx as usize) {
            *slot = true;
        }
    }
    seen.into_iter()
        .enumerate()
        .filter_map(|(idx, seen)| (!seen).then_some(idx as u32))
        .collect()
}

fn concat_same_schema(mut frames: Vec<DataFrame>) -> Result<DataFrame> {
    let mut iter = frames.drain(..);
    let mut out = iter.next().unwrap_or_else(DataFrame::empty);
    for frame in iter {
        out.vstack_mut(&frame)?;
    }
    Ok(out)
}

fn append_overlap_length_column(
    df: &mut DataFrame,
    column_name: &str,
    start_col: &str,
    end_col: &str,
    suffix: &str,
) -> Result<()> {
    let right_start = format!("{start_col}{suffix}");
    let right_end = format!("{end_col}{suffix}");
    let left_starts = integer_column_to_i64(df, start_col)?;
    let left_ends = integer_column_to_i64(df, end_col)?;
    let right_starts = integer_column_to_i64(df, &right_start)?;
    let right_ends = integer_column_to_i64(df, &right_end)?;
    let overlap = left_starts
        .into_iter()
        .zip(left_ends)
        .zip(right_starts)
        .zip(right_ends)
        .map(|(((left_start, left_end), right_start), right_end)| {
            left_end.min(right_end) - left_start.max(right_start)
        })
        .collect::<Vec<_>>();
    df.with_column(Column::from(Series::new(column_name.into(), overlap)))?;
    Ok(())
}

fn append_null_overlap_length_column(
    df: &mut DataFrame,
    options: &JoinOverlapsOptions,
    len: usize,
) -> Result<()> {
    if let Some(column_name) = options.report_overlap_column.as_deref() {
        df.with_column(Column::full_null(column_name.into(), len, &DataType::Int64))?;
    }
    Ok(())
}

fn integer_column_to_i64(df: &DataFrame, column_name: &str) -> Result<Vec<i64>> {
    let column = df.column(column_name)?;
    if column.null_count() > 0 {
        return Err(RangeFrameError::NullValues {
            column: column_name.to_owned(),
        });
    }

    let series = column.as_materialized_series();
    let casted = match series.dtype() {
        DataType::Int8
        | DataType::Int16
        | DataType::Int32
        | DataType::Int64
        | DataType::UInt8
        | DataType::UInt16
        | DataType::UInt32
        | DataType::UInt64 => series.cast(&DataType::Int64)?,
        dtype => {
            return Err(RangeFrameError::InvalidCoordinateDtype {
                column: column_name.to_owned(),
                dtype: dtype.to_string(),
            });
        }
    };
    Ok(casted.i64()?.into_no_null_iter().collect())
}

fn complement_lengths_i32(
    df: &DataFrame,
    groups: &[u32],
    options: &ComplementRangesOptions,
) -> Result<FxHashMap<u32, i32>> {
    let mut out = FxHashMap::default();
    let Some(chromsizes) = options.chromsizes.as_ref() else {
        return Ok(out);
    };

    let col = df.column(&options.group_sizes_col)?;
    for (row_idx, group) in groups.iter().copied().enumerate() {
        if out.contains_key(&group) {
            continue;
        }
        let key = any_value_key(col.get(row_idx)?);
        if let Some(length) = chromsizes.get(&key) {
            if let Ok(length) = i32::try_from(*length) {
                out.insert(group, length);
            }
        }
    }
    Ok(out)
}

fn complement_lengths_i64(
    df: &DataFrame,
    groups: &[u32],
    options: &ComplementRangesOptions,
) -> Result<FxHashMap<u32, i64>> {
    let mut out = FxHashMap::default();
    let Some(chromsizes) = options.chromsizes.as_ref() else {
        return Ok(out);
    };

    let col = df.column(&options.group_sizes_col)?;
    for (row_idx, group) in groups.iter().copied().enumerate() {
        if out.contains_key(&group) {
            continue;
        }
        let key = any_value_key(col.get(row_idx)?);
        if let Some(length) = chromsizes.get(&key) {
            out.insert(group, *length);
        }
    }
    Ok(out)
}

fn any_value_key(value: AnyValue<'_>) -> String {
    match value {
        AnyValue::String(value) => value.to_owned(),
        AnyValue::StringOwned(value) => value.as_str().to_owned(),
        _ => value.to_string(),
    }
}

fn all_indices(len: usize) -> Result<Vec<u32>> {
    if len > u32::MAX as usize {
        return Err(RangeFrameError::TooManyRows { len });
    }
    Ok((0..len as u32).collect())
}

fn usize_indices_to_u32(indices: &[usize]) -> Result<Vec<u32>> {
    indices
        .iter()
        .map(|&idx| u32::try_from(idx).map_err(|_| RangeFrameError::TooManyRows { len: idx }))
        .collect()
}

fn resolve_extend_options(options: &ExtendRangesOptions) -> Result<(i64, i64)> {
    let has_ext = options.ext.is_some();
    let has_ends = options.ext_3.is_some() || options.ext_5.is_some();
    if has_ext == has_ends {
        return Err(compute_error(
            "must use at least one and not both of ext and ext_3/ext_5",
        ));
    }

    if let Some(ext) = options.ext {
        Ok((ext, ext))
    } else {
        Ok((options.ext_3.unwrap_or(0), options.ext_5.unwrap_or(0)))
    }
}

fn negative_strand_flags(df: &DataFrame, column_name: &str) -> Result<Vec<bool>> {
    let column = df
        .column(column_name)
        .map_err(|_| RangeFrameError::MissingStrandColumn {
            column: column_name.to_owned(),
        })?;
    if column.null_count() > 0 {
        return Err(RangeFrameError::NullValues {
            column: column_name.to_owned(),
        });
    }

    let values = column.as_materialized_series().str().map_err(|_| {
        RangeFrameError::UnsupportedKeyDtype {
            column: column_name.to_owned(),
            dtype: column.dtype().to_string(),
        }
    })?;
    Ok(values
        .into_no_null_iter()
        .map(|value| value == "-")
        .collect())
}

fn forward_strand_flags(df: &DataFrame, column_name: &str) -> Result<Vec<bool>> {
    Ok(negative_strand_flags(df, column_name)?
        .into_iter()
        .map(|is_negative| !is_negative)
        .collect())
}

fn row_aligned_bounds_i32(
    df: &DataFrame,
    group_col: &str,
    ends: &[i32],
    options: &ClipRangesOptions,
) -> Result<Vec<i32>> {
    let bounds = row_aligned_bounds_i64(
        df,
        group_col,
        &ends.iter().map(|&v| v as i64).collect::<Vec<_>>(),
        options,
    )?;
    bounds
        .into_iter()
        .map(|value| i32::try_from(value).map_err(|_| compute_error("chromsize exceeds i32 range")))
        .collect()
}

fn row_aligned_bounds_i64(
    df: &DataFrame,
    group_col: &str,
    ends: &[i64],
    options: &ClipRangesOptions,
) -> Result<Vec<i64>> {
    if options.chromsizes.is_none() {
        let max_end = ends.iter().copied().max().unwrap_or(0);
        return Ok(vec![max_end; df.height()]);
    }

    let chromsizes = options.chromsizes.as_ref().expect("checked above");
    let col = df.column(group_col)?;
    let mut out = Vec::with_capacity(df.height());
    let mut missing = HashSet::new();
    for row_idx in 0..df.height() {
        let key = any_value_key(col.get(row_idx)?);
        if let Some(length) = chromsizes.get(&key) {
            out.push(*length);
        } else {
            missing.insert(key);
            out.push(0);
        }
    }

    if !missing.is_empty() {
        return Err(compute_error(format!(
            "not all groups were in the chromsize map; missing keys: {missing:?}"
        )));
    }

    Ok(out)
}

fn append_or_replace_cumsum_i32(
    df: &mut DataFrame,
    start_col: &str,
    end_col: &str,
    starts: Vec<i32>,
    ends: Vec<i32>,
    options: &GroupCumsumOptions,
) -> Result<()> {
    match (
        options.cumsum_start_column.as_deref(),
        options.cumsum_end_column.as_deref(),
    ) {
        (Some(start_name), Some(end_name)) => {
            df.with_column(Column::from(Series::new(start_name.into(), starts)))?;
            df.with_column(Column::from(Series::new(end_name.into(), ends)))?;
        }
        (None, None) => replace_interval_columns_i32(df, start_col, end_col, starts, ends)?,
        _ => {
            return Err(compute_error(
                "both cumsum_start_column and cumsum_end_column must be set",
            ))
        }
    }
    Ok(())
}

fn append_or_replace_cumsum_i64(
    df: &mut DataFrame,
    start_col: &str,
    end_col: &str,
    starts: Vec<i64>,
    ends: Vec<i64>,
    options: &GroupCumsumOptions,
) -> Result<()> {
    match (
        options.cumsum_start_column.as_deref(),
        options.cumsum_end_column.as_deref(),
    ) {
        (Some(start_name), Some(end_name)) => {
            df.with_column(Column::from(Series::new(start_name.into(), starts)))?;
            df.with_column(Column::from(Series::new(end_name.into(), ends)))?;
        }
        (None, None) => replace_interval_columns_i64(df, start_col, end_col, starts, ends)?,
        _ => {
            return Err(compute_error(
                "both cumsum_start_column and cumsum_end_column must be set",
            ))
        }
    }
    Ok(())
}

fn compute_error(message: impl Into<String>) -> RangeFrameError {
    RangeFrameError::Polars(polars_core::prelude::PolarsError::ComputeError(
        message.into().into(),
    ))
}

fn replace_interval_columns_i32(
    df: &mut DataFrame,
    start_col: &str,
    end_col: &str,
    starts: Vec<i32>,
    ends: Vec<i32>,
) -> Result<()> {
    df.with_column(Column::from(Series::new(start_col.into(), starts)))?;
    df.with_column(Column::from(Series::new(end_col.into(), ends)))?;
    Ok(())
}

fn replace_interval_columns_i64(
    df: &mut DataFrame,
    start_col: &str,
    end_col: &str,
    starts: Vec<i64>,
    ends: Vec<i64>,
) -> Result<()> {
    df.with_column(Column::from(Series::new(start_col.into(), starts)))?;
    df.with_column(Column::from(Series::new(end_col.into(), ends)))?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use polars_core::prelude::{Column, DataFrame, NamedFrom, Series};
    use rustc_hash::FxHashMap;

    use super::{
        ClipRangesOptions, ClusterOptions, ComplementRangesOptions, CountOverlapsOptions,
        ExtendRangesOptions, GroupCumsumOptions, IntersectOverlapsOptions, JoinOverlapsOptions,
        JoinType, MaxDisjointOptions, MergeOptions, NearestDirection, NearestOptions,
        OuterRangesOptions, SetIntersectOverlapsOptions, SetUnionOverlapsOptions,
        SortRangesOptions, SplitOverlapsOptions, SubtractOptions, TileRangesOptions,
    };
    use crate::dataframe_ext::DataFrameRanges;

    fn make_df(columns: Vec<Series>) -> DataFrame {
        let columns = columns.into_iter().map(Column::from).collect::<Vec<_>>();
        DataFrame::new_infer_height(columns).unwrap()
    }

    #[test]
    fn merge_overlaps_matches_pyranges_style_projection_and_counts() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr2"]),
            Series::new("Start".into(), &[1_i64, 4, 10]),
            Series::new("End".into(), &[5_i64, 8, 12]),
            Series::new("Score".into(), &[1_i32, 2, 3]),
        ]);

        let result = df
            .merge_overlaps(MergeOptions {
                count_column: Some("Count".to_owned()),
                ..MergeOptions::default().with_match_by(["Chrom"])
            })
            .unwrap();

        assert_eq!(
            result.get_column_names(),
            &["Chrom", "Start", "End", "Count"]
        );
        assert_eq!(result.height(), 2);
        assert_eq!(
            result
                .column("Count")
                .unwrap()
                .u32()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![2, 1]
        );
    }

    #[test]
    fn cluster_overlaps_adds_cluster_column() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1", "chr2"]),
            Series::new("Start".into(), &[1_i64, 3, 10, 1]),
            Series::new("End".into(), &[4_i64, 5, 12, 2]),
        ]);

        let result = df
            .cluster_overlaps(ClusterOptions::default().with_match_by(["Chrom"]))
            .unwrap();

        assert_eq!(result.height(), 4);
        assert_eq!(
            result
                .column("Cluster")
                .unwrap()
                .u32()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![0, 0, 1, 3]
        );
    }

    #[test]
    fn nearest_ranges_appends_suffixed_columns_and_distance() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 20]),
            Series::new("End".into(), &[5_i64, 25]),
            Series::new("Name".into(), &["a", "b"]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[8_i64, 30]),
            Series::new("End".into(), &[10_i64, 35]),
            Series::new("Name".into(), &["x", "y"]),
        ]);

        let result = left
            .nearest_ranges(
                &right,
                NearestOptions {
                    direction: NearestDirection::Forward,
                    ..NearestOptions::default().with_match_by(["Chrom"])
                },
            )
            .unwrap();

        assert_eq!(result.height(), 2);
        assert!(result.column("Chrom_b").is_ok());
        assert!(result.column("Distance").is_ok());
    }

    #[test]
    fn subtract_overlaps_splits_intervals_and_preserves_other_columns() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1"]),
            Series::new("Start".into(), &[1_i64]),
            Series::new("End".into(), &[10_i64]),
            Series::new("Name".into(), &["a"]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1"]),
            Series::new("Start".into(), &[4_i64]),
            Series::new("End".into(), &[6_i64]),
        ]);

        let result = left
            .subtract_overlaps(&right, SubtractOptions::default().with_match_by(["Chrom"]))
            .unwrap();

        assert_eq!(result.height(), 2);
        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![1, 6]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![4, 10]
        );
    }

    #[test]
    fn count_overlaps_returns_one_count_per_left_interval() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 10, 30]),
            Series::new("End".into(), &[5_i64, 20, 40]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[3_i64, 11, 18, 35]),
            Series::new("End".into(), &[4_i64, 12, 19, 36]),
        ]);

        let counts = left
            .count_overlaps(
                &right,
                CountOverlapsOptions::default().with_match_by(["Chrom"]),
            )
            .unwrap();

        assert_eq!(
            counts
                .u32()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![1, 2, 1]
        );
    }

    #[test]
    fn join_overlaps_suffixes_right_collisions_and_supports_left_join() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 10]),
            Series::new("End".into(), &[5_i64, 20]),
            Series::new("Name".into(), &["a", "b"]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1"]),
            Series::new("Start".into(), &[3_i64]),
            Series::new("End".into(), &[4_i64]),
            Series::new("Score".into(), &[7_i32]),
        ]);

        let result = left
            .join_overlaps(
                &right,
                JoinOverlapsOptions {
                    join_type: JoinType::Left,
                    ..JoinOverlapsOptions::default().with_match_by(["Chrom"])
                },
            )
            .unwrap();

        assert_eq!(result.height(), 2);
        assert!(result.column("Start_b").is_ok());
        assert!(result.column("Score").is_ok());
        assert_eq!(result.column("Score").unwrap().null_count(), 1);
    }

    #[test]
    fn join_overlaps_left_join_keeps_report_column_for_unmatched_rows() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 20]),
            Series::new("End".into(), &[5_i64, 25]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1"]),
            Series::new("Start".into(), &[3_i64]),
            Series::new("End".into(), &[7_i64]),
        ]);

        let result = left
            .join_overlaps(
                &right,
                JoinOverlapsOptions {
                    join_type: JoinType::Left,
                    report_overlap_column: Some("Overlap".to_owned()),
                    ..JoinOverlapsOptions::default().with_match_by(["Chrom"])
                },
            )
            .unwrap();

        assert_eq!(result.height(), 2);
        let overlap = result.column("Overlap").unwrap().i64().unwrap();
        assert_eq!(overlap.get(0), Some(2));
        assert_eq!(overlap.get(1), None);
    }

    #[test]
    fn intersect_overlaps_replaces_coordinates_with_intersection() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1"]),
            Series::new("Start".into(), &[1_i64]),
            Series::new("End".into(), &[10_i64]),
            Series::new("Name".into(), &["a"]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1"]),
            Series::new("Start".into(), &[4_i64]),
            Series::new("End".into(), &[6_i64]),
        ]);

        let result = left
            .intersect_overlaps(
                &right,
                IntersectOverlapsOptions::default().with_match_by(["Chrom"]),
            )
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![4]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![6]
        );
    }

    #[test]
    fn set_intersect_overlaps_merges_inputs_first() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[5_i64, 20, 40]),
            Series::new("End".into(), &[10_i64, 30, 50]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[7_i64, 18, 25, 28]),
            Series::new("End".into(), &[9_i64, 22, 33, 32]),
        ]);

        let result = left
            .set_intersect_overlaps(
                &right,
                SetIntersectOverlapsOptions::default().with_match_by(["Chrom"]),
            )
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![7, 20, 25]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![9, 22, 30]
        );
    }

    #[test]
    fn set_union_overlaps_projects_and_merges_both_inputs() {
        let left = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 4, 10]),
            Series::new("End".into(), &[3_i64, 9, 11]),
            Series::new("Name".into(), &["a", "b", "c"]),
        ]);
        let right = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[2_i64, 2, 9]),
            Series::new("End".into(), &[3_i64, 9, 10]),
        ]);

        let result = left
            .set_union_overlaps(
                &right,
                SetUnionOverlapsOptions::default().with_match_by(["Chrom"]),
            )
            .unwrap();

        assert_eq!(result.get_column_names(), &["Chrom", "Start", "End"]);
        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![1, 9, 10]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![9, 10, 11]
        );
    }

    #[test]
    fn sort_ranges_uses_factorized_groups_and_reverse_strand_order() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1", "chr1"]),
            Series::new("Strand".into(), &["+", "+", "-", "-"]),
            Series::new("Start".into(), &[40_i64, 1, 10, 70]),
            Series::new("End".into(), &[60_i64, 11, 25, 80]),
        ]);

        let result = df
            .sort_ranges(SortRangesOptions {
                match_by: vec!["Chrom".to_owned(), "Strand".to_owned()],
                negative_strand_column: Some("Strand".to_owned()),
            })
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![1, 40, 70, 10]
        );
    }

    #[test]
    fn extend_ranges_supports_per_row_and_grouped_extensions() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[3_i64, 8, 5]),
            Series::new("End".into(), &[6_i64, 9, 7]),
            Series::new("Strand".into(), &["+", "+", "-"]),
            Series::new("Tx".into(), &["a", "a", "b"]),
        ]);

        let per_row = df
            .extend_ranges(ExtendRangesOptions {
                ext: Some(3),
                ..ExtendRangesOptions::default()
            })
            .unwrap();
        assert_eq!(
            per_row
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![0, 5, 2]
        );

        let grouped = df
            .extend_ranges(ExtendRangesOptions {
                match_by: vec!["Tx".to_owned()],
                ext_3: Some(3),
                ext_5: Some(0),
                negative_strand_column: Some("Strand".to_owned()),
                ..ExtendRangesOptions::default()
            })
            .unwrap();
        assert_eq!(
            grouped
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![6, 12, 7]
        );
        assert_eq!(
            grouped
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![3, 8, 2]
        );
    }

    #[test]
    fn tile_ranges_emits_tiles_and_overlap_fraction() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[99_i64, 100]),
            Series::new("End".into(), &[100_i64, 250]),
        ]);

        let result = df
            .tile_ranges(TileRangesOptions {
                overlap_column: Some("TileOverlap".to_owned()),
                ..TileRangesOptions::new(100).with_match_by(["Chrom"])
            })
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![0, 100, 200]
        );
        assert_eq!(
            result
                .column("TileOverlap")
                .unwrap()
                .f64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![0.01, 1.0, 0.5]
        );
    }

    #[test]
    fn clip_ranges_clips_or_removes_out_of_bounds_intervals() {
        let df = make_df(vec![
            Series::new("Chromosome".into(), &["1", "1", "3"]),
            Series::new("Start".into(), &[1_i64, 249250600, 5]),
            Series::new("End".into(), &[2_i64, 249250640, 7]),
        ]);
        let mut chromsizes = FxHashMap::default();
        chromsizes.insert("1".to_owned(), 249250621);
        chromsizes.insert("3".to_owned(), 500);

        let clipped = df
            .clip_ranges(ClipRangesOptions {
                chromsizes: Some(chromsizes.clone()),
                ..ClipRangesOptions::default()
            })
            .unwrap();
        assert_eq!(
            clipped
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![2, 249250621, 7]
        );

        let removed = df
            .clip_ranges(ClipRangesOptions {
                chromsizes: Some(chromsizes),
                remove: true,
                ..ClipRangesOptions::default()
            })
            .unwrap();
        assert_eq!(removed.height(), 2);
    }

    #[test]
    fn group_cumsum_adds_running_coordinates_per_group() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 40, 70]),
            Series::new("End".into(), &[11_i64, 60, 80]),
            Series::new("Tx".into(), &["t1", "t1", "t2"]),
        ]);

        let result = df
            .group_cumsum(GroupCumsumOptions {
                match_by: vec!["Chrom".to_owned(), "Tx".to_owned()],
                cumsum_start_column: Some("CStart".to_owned()),
                cumsum_end_column: Some("CEnd".to_owned()),
                ..GroupCumsumOptions::default()
            })
            .unwrap();

        assert_eq!(
            result
                .column("CStart")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![0, 10, 0]
        );
        assert_eq!(
            result
                .column("CEnd")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![10, 30, 10]
        );
    }

    #[test]
    fn max_disjoint_overlaps_selects_non_overlapping_subset() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 4, 10]),
            Series::new("End".into(), &[5_i64, 7, 14]),
        ]);

        let result = df
            .max_disjoint_overlaps(MaxDisjointOptions::default().with_match_by(["Chrom"]))
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![1, 10]
        );
    }

    #[test]
    fn split_overlaps_emits_atomic_segments() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[3_i64, 5]),
            Series::new("End".into(), &[6_i64, 9]),
        ]);

        let result = df
            .split_overlaps(SplitOverlapsOptions::default().with_match_by(["Chrom"]))
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![3, 5, 6]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![5, 6, 9]
        );
    }

    #[test]
    fn outer_ranges_returns_group_boundaries() {
        let df = make_df(vec![
            Series::new("Transcript".into(), &["tr1", "tr1", "tr2"]),
            Series::new("Start".into(), &[1_i64, 60, 110]),
            Series::new("End".into(), &[40_i64, 68, 130]),
        ]);

        let result = df
            .outer_ranges(OuterRangesOptions::default().with_match_by(["Transcript"]))
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![1, 110]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![68, 130]
        );
    }

    #[test]
    fn complement_ranges_returns_internal_gaps_after_merge() {
        let df = make_df(vec![
            Series::new("Chrom".into(), &["chr1", "chr1", "chr1"]),
            Series::new("Start".into(), &[2_i64, 10, 12]),
            Series::new("End".into(), &[5_i64, 11, 18]),
        ]);

        let result = df
            .complement_ranges(ComplementRangesOptions::default().with_match_by(["Chrom"]))
            .unwrap();

        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![5, 11]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![10, 12]
        );
    }
}