polaranges 0.3.2

use std::borrow::Cow;

use polars_core::prelude::{Column, DataFrame, NamedFrom, Series};

use crate::dataframe_ext::{overlap_indices_fast, DataFrameRanges};
use crate::factorize::borrowed_string_values;
use crate::range_frame::{take_rows, take_rows_owned};
use crate::{
    ClipRangesOptions, ClusterOptions, ComplementRangesOptions, CountOverlapsOptions,
    ExtendRangesOptions, GroupCumsumOptions, IntersectOverlapsOptions, JoinOverlapsOptions,
    JoinType, MaxDisjointOptions, MergeOptions, NearestDirection, NearestOptions,
    OuterRangesOptions, OverlapMode, OverlapOptions, RangeFrameError, Result,
    SetIntersectOverlapsOptions, SetUnionOverlapsOptions, SortRangesOptions, SplitOverlapsOptions,
    SubtractOptions, TileRangesOptions,
};

const DEFAULT_CHROMOSOME_COL: &str = "Chromosome";
const DEFAULT_STRAND_COL: &str = "Strand";
const POSITIVE_STRAND: &str = "+";
const NEGATIVE_STRAND: &str = "-";
const TEMP_ROW_ID_PREFIX: &str = "__polaranges_rowid__";

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum UseStrand {
    Auto,
    Enabled,
    Disabled,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StrandBehavior {
    Auto,
    Same,
    Opposite,
    Ignore,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BioNearestDirection {
    Any,
    Upstream,
    Downstream,
}

#[derive(Clone, Debug)]
pub struct BioMergeOptions {
    pub use_strand: UseStrand,
    pub count_column: Option<String>,
    pub match_by: Vec<String>,
    pub slack: i64,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioMergeOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            count_column: None,
            match_by: Vec::new(),
            slack: 0,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioMergeOptions {
    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 BioClusterOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub cluster_column: String,
    pub slack: i64,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioClusterOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            cluster_column: "Cluster".to_owned(),
            slack: 0,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioClusterOptions {
    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 BioOverlapOptions {
    pub multiple: OverlapMode,
    pub slack: i64,
    pub contained_intervals_only: bool,
    pub match_by: Vec<String>,
    pub preserve_input_order: bool,
    pub strand_behavior: StrandBehavior,
    pub invert: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioOverlapOptions {
    fn default() -> Self {
        Self {
            multiple: OverlapMode::First,
            slack: 0,
            contained_intervals_only: false,
            match_by: Vec::new(),
            preserve_input_order: true,
            strand_behavior: StrandBehavior::Auto,
            invert: false,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioOverlapOptions {
    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 BioNearestOptions {
    pub strand_behavior: StrandBehavior,
    pub match_by: Vec<String>,
    pub suffix: String,
    pub exclude_overlaps: bool,
    pub k: usize,
    pub distance_column: Option<String>,
    pub direction: BioNearestDirection,
    pub preserve_input_order: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioNearestOptions {
    fn default() -> Self {
        Self {
            strand_behavior: StrandBehavior::Auto,
            match_by: Vec::new(),
            suffix: "_b".to_owned(),
            exclude_overlaps: false,
            k: 1,
            distance_column: Some("Distance".to_owned()),
            direction: BioNearestDirection::Any,
            preserve_input_order: true,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioNearestOptions {
    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 BioSubtractOptions {
    pub strand_behavior: StrandBehavior,
    pub match_by: Vec<String>,
    pub preserve_input_order: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioSubtractOptions {
    fn default() -> Self {
        Self {
            strand_behavior: StrandBehavior::Auto,
            match_by: Vec::new(),
            preserve_input_order: true,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioSubtractOptions {
    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 BioCountOverlapsOptions {
    pub strand_behavior: StrandBehavior,
    pub match_by: Vec<String>,
    pub slack: i64,
    pub overlap_column: String,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioCountOverlapsOptions {
    fn default() -> Self {
        Self {
            strand_behavior: StrandBehavior::Auto,
            match_by: Vec::new(),
            slack: 0,
            overlap_column: "Count".to_owned(),
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioCountOverlapsOptions {
    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 BioJoinOverlapsOptions {
    pub strand_behavior: StrandBehavior,
    pub match_by: Vec<String>,
    pub multiple: 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,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioJoinOverlapsOptions {
    fn default() -> Self {
        Self {
            strand_behavior: StrandBehavior::Auto,
            match_by: Vec::new(),
            multiple: OverlapMode::All,
            slack: 0,
            suffix: "_b".to_owned(),
            contained_intervals_only: false,
            join_type: JoinType::Inner,
            report_overlap_column: None,
            preserve_input_order: true,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioJoinOverlapsOptions {
    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 BioIntersectOverlapsOptions {
    pub strand_behavior: StrandBehavior,
    pub match_by: Vec<String>,
    pub multiple: OverlapMode,
    pub preserve_input_order: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioIntersectOverlapsOptions {
    fn default() -> Self {
        Self {
            strand_behavior: StrandBehavior::Auto,
            match_by: Vec::new(),
            multiple: OverlapMode::All,
            preserve_input_order: true,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioSetIntersectOverlapsOptions {
    pub strand_behavior: StrandBehavior,
    pub match_by: Vec<String>,
    pub multiple: OverlapMode,
    pub preserve_input_order: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioSetIntersectOverlapsOptions {
    fn default() -> Self {
        Self {
            strand_behavior: StrandBehavior::Auto,
            match_by: Vec::new(),
            multiple: OverlapMode::All,
            preserve_input_order: true,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioSetUnionOverlapsOptions {
    pub strand_behavior: StrandBehavior,
    pub match_by: Vec<String>,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioSetUnionOverlapsOptions {
    fn default() -> Self {
        Self {
            strand_behavior: StrandBehavior::Auto,
            match_by: Vec::new(),
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioSortOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioSortOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioExtendOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub ext: Option<i64>,
    pub ext_3: Option<i64>,
    pub ext_5: Option<i64>,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioExtendOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            ext: None,
            ext_3: None,
            ext_5: None,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioExtendOptions {
    pub fn new(ext: i64) -> Self {
        Self {
            ext: Some(ext),
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
            ..Self::default()
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioTileOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub tile_size: i64,
    pub overlap_column: Option<String>,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl BioTileOptions {
    pub fn new(tile_size: i64) -> Self {
        Self {
            use_strand: UseStrand::Disabled,
            match_by: Vec::new(),
            tile_size,
            overlap_column: None,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

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

impl Default for BioClipOptions {
    fn default() -> Self {
        Self {
            chromsizes: None,
            remove: false,
            only_right: false,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioGroupCumsumOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub cumsum_start_column: Option<String>,
    pub cumsum_end_column: Option<String>,
    pub keep_order: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioGroupCumsumOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            cumsum_start_column: None,
            cumsum_end_column: None,
            keep_order: true,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioMaxDisjointOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub slack: i64,
    pub preserve_input_order: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioMaxDisjointOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            slack: 0,
            preserve_input_order: true,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

impl BioMaxDisjointOptions {
    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 BioSplitOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub between: bool,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioSplitOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            between: false,
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioOuterRangesOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioOuterRangesOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct BioComplementRangesOptions {
    pub use_strand: UseStrand,
    pub match_by: Vec<String>,
    pub include_first_interval: bool,
    pub chromsizes: Option<rustc_hash::FxHashMap<String, i64>>,
    pub group_sizes_col: String,
    pub chromosome_column: String,
    pub strand_column: String,
}

impl Default for BioComplementRangesOptions {
    fn default() -> Self {
        Self {
            use_strand: UseStrand::Auto,
            match_by: Vec::new(),
            include_first_interval: false,
            chromsizes: None,
            group_sizes_col: DEFAULT_CHROMOSOME_COL.to_owned(),
            chromosome_column: DEFAULT_CHROMOSOME_COL.to_owned(),
            strand_column: DEFAULT_STRAND_COL.to_owned(),
        }
    }
}

pub(crate) trait BioDataFrameRanges {
    fn bio_merge_overlaps(&self, options: BioMergeOptions) -> Result<DataFrame>;
    fn bio_cluster_overlaps(&self, options: BioClusterOptions) -> Result<DataFrame>;
    fn bio_count_overlaps(
        &self,
        other: &DataFrame,
        options: BioCountOverlapsOptions,
    ) -> Result<DataFrame>;
    fn bio_join_overlaps(
        &self,
        other: &DataFrame,
        options: BioJoinOverlapsOptions,
    ) -> Result<DataFrame>;
    fn bio_intersect_overlaps(
        &self,
        other: &DataFrame,
        options: BioIntersectOverlapsOptions,
    ) -> Result<DataFrame>;
    fn bio_set_intersect_overlaps(
        &self,
        other: &DataFrame,
        options: BioSetIntersectOverlapsOptions,
    ) -> Result<DataFrame>;
    fn bio_set_union_overlaps(
        &self,
        other: &DataFrame,
        options: BioSetUnionOverlapsOptions,
    ) -> Result<DataFrame>;
    fn bio_sort_ranges(&self, options: BioSortOptions) -> Result<DataFrame>;
    fn bio_extend_ranges(&self, options: BioExtendOptions) -> Result<DataFrame>;
    fn bio_tile_ranges(&self, options: BioTileOptions) -> Result<DataFrame>;
    fn bio_clip_ranges(&self, options: BioClipOptions) -> Result<DataFrame>;
    fn bio_group_cumsum(&self, options: BioGroupCumsumOptions) -> Result<DataFrame>;
    fn bio_max_disjoint_overlaps(&self, options: BioMaxDisjointOptions) -> Result<DataFrame>;
    fn bio_split_overlaps(&self, options: BioSplitOptions) -> Result<DataFrame>;
    fn bio_outer_ranges(&self, options: BioOuterRangesOptions) -> Result<DataFrame>;
    fn bio_complement_ranges(&self, options: BioComplementRangesOptions) -> Result<DataFrame>;
    fn bio_overlap_ranges(
        &self,
        other: &DataFrame,
        options: BioOverlapOptions,
    ) -> Result<DataFrame>;
    fn bio_nearest_ranges(
        &self,
        other: &DataFrame,
        options: BioNearestOptions,
    ) -> Result<DataFrame>;
    fn bio_subtract_overlaps(
        &self,
        other: &DataFrame,
        options: BioSubtractOptions,
    ) -> Result<DataFrame>;
    fn bio_has_valid_strand(&self, strand_column: Option<&str>) -> bool;
}

impl BioDataFrameRanges for DataFrame {
    fn bio_merge_overlaps(&self, options: BioMergeOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;

        self.merge_overlaps(MergeOptions {
            count_column: options.count_column,
            match_by,
            slack: options.slack,
        })
    }

    fn bio_cluster_overlaps(&self, options: BioClusterOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;

        self.cluster_overlaps(ClusterOptions {
            match_by,
            cluster_column: options.cluster_column,
            slack: options.slack,
        })
    }

    fn bio_count_overlaps(
        &self,
        other: &DataFrame,
        options: BioCountOverlapsOptions,
    ) -> Result<DataFrame> {
        let (prepared_other, match_by) = prepare_binary_match_by(
            self,
            other,
            options.strand_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        let mut out = self.clone();
        let mut counts = self.count_overlaps(
            prepared_other.as_ref(),
            CountOverlapsOptions {
                match_by,
                slack: options.slack,
            },
        )?;
        counts.rename(options.overlap_column.into());
        out.with_column(Column::from(counts))?;
        Ok(out)
    }

    fn bio_join_overlaps(
        &self,
        other: &DataFrame,
        options: BioJoinOverlapsOptions,
    ) -> Result<DataFrame> {
        let (prepared_other, match_by) = prepare_binary_match_by(
            self,
            other,
            options.strand_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.join_overlaps(
            prepared_other.as_ref(),
            JoinOverlapsOptions {
                match_by,
                multiple: options.multiple,
                slack: options.slack,
                suffix: options.suffix,
                contained_intervals_only: options.contained_intervals_only,
                join_type: options.join_type,
                report_overlap_column: options.report_overlap_column,
                preserve_input_order: options.preserve_input_order,
            },
        )
    }

    fn bio_intersect_overlaps(
        &self,
        other: &DataFrame,
        options: BioIntersectOverlapsOptions,
    ) -> Result<DataFrame> {
        let (prepared_other, match_by) = prepare_binary_match_by(
            self,
            other,
            options.strand_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.intersect_overlaps(
            prepared_other.as_ref(),
            IntersectOverlapsOptions {
                match_by,
                multiple: options.multiple,
                slack: 0,
                contained_intervals_only: false,
                preserve_input_order: options.preserve_input_order,
            },
        )
    }

    fn bio_set_intersect_overlaps(
        &self,
        other: &DataFrame,
        options: BioSetIntersectOverlapsOptions,
    ) -> Result<DataFrame> {
        let (prepared_other, match_by) = prepare_binary_match_by(
            self,
            other,
            options.strand_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.set_intersect_overlaps(
            prepared_other.as_ref(),
            SetIntersectOverlapsOptions {
                match_by,
                multiple: options.multiple,
                preserve_input_order: options.preserve_input_order,
            },
        )
    }

    fn bio_set_union_overlaps(
        &self,
        other: &DataFrame,
        options: BioSetUnionOverlapsOptions,
    ) -> Result<DataFrame> {
        let (prepared_other, match_by) = prepare_binary_match_by(
            self,
            other,
            options.strand_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.set_union_overlaps(
            prepared_other.as_ref(),
            SetUnionOverlapsOptions { match_by },
        )
    }

    fn bio_sort_ranges(&self, options: BioSortOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        let negative_strand_column = match_by
            .iter()
            .any(|column| column == &options.strand_column)
            .then_some(options.strand_column);
        self.sort_ranges(SortRangesOptions {
            match_by,
            negative_strand_column,
        })
    }

    fn bio_extend_ranges(&self, options: BioExtendOptions) -> Result<DataFrame> {
        let use_strand = use_strand_enabled(self, options.use_strand, &options.strand_column)?;
        self.extend_ranges(ExtendRangesOptions {
            match_by: options.match_by,
            ext: options.ext,
            ext_3: options.ext_3,
            ext_5: options.ext_5,
            negative_strand_column: use_strand.then_some(options.strand_column),
        })
    }

    fn bio_tile_ranges(&self, options: BioTileOptions) -> Result<DataFrame> {
        let use_strand = use_strand_enabled(self, options.use_strand, &options.strand_column)?;
        self.tile_ranges(TileRangesOptions {
            match_by: options.match_by,
            tile_size: options.tile_size,
            negative_strand_column: use_strand.then_some(options.strand_column),
            overlap_column: options.overlap_column,
        })
    }

    fn bio_clip_ranges(&self, options: BioClipOptions) -> Result<DataFrame> {
        self.clip_ranges(ClipRangesOptions {
            chromsizes: options.chromsizes,
            remove: options.remove,
            only_right: options.only_right,
            group_sizes_col: options.chromosome_column,
        })
    }

    fn bio_group_cumsum(&self, options: BioGroupCumsumOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        let forward_strand_column = match_by
            .iter()
            .any(|column| column == &options.strand_column)
            .then_some(options.strand_column);
        self.group_cumsum(GroupCumsumOptions {
            match_by,
            forward_strand_column,
            cumsum_start_column: options.cumsum_start_column,
            cumsum_end_column: options.cumsum_end_column,
            keep_order: options.keep_order,
        })
    }

    fn bio_max_disjoint_overlaps(&self, options: BioMaxDisjointOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.max_disjoint_overlaps(MaxDisjointOptions {
            match_by,
            slack: options.slack,
            preserve_input_order: options.preserve_input_order,
        })
    }

    fn bio_split_overlaps(&self, options: BioSplitOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.split_overlaps(SplitOverlapsOptions {
            match_by,
            between: options.between,
        })
    }

    fn bio_outer_ranges(&self, options: BioOuterRangesOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.outer_ranges(OuterRangesOptions { match_by })
    }

    fn bio_complement_ranges(&self, options: BioComplementRangesOptions) -> Result<DataFrame> {
        let match_by = prepare_single_match_by(
            self,
            options.use_strand,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        self.complement_ranges(ComplementRangesOptions {
            match_by,
            include_first_interval: options.include_first_interval,
            chromsizes: options.chromsizes,
            group_sizes_col: options.group_sizes_col,
        })
    }

    fn bio_overlap_ranges(
        &self,
        other: &DataFrame,
        options: BioOverlapOptions,
    ) -> Result<DataFrame> {
        let (prepared_other, match_by) = prepare_binary_match_by(
            self,
            other,
            options.strand_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;

        let overlap_options = OverlapOptions {
            multiple: options.multiple,
            slack: options.slack,
            contained_intervals_only: options.contained_intervals_only,
            match_by,
            preserve_input_order: options.preserve_input_order,
            ..OverlapOptions::default()
        };

        if options.invert {
            let overlapping_rows = overlap_indices_fast(
                self,
                prepared_other.as_ref(),
                &overlap_options,
            )?;
            return take_non_overlapping_rows(self, overlapping_rows);
        }

        self.range_overlap(prepared_other.as_ref(), overlap_options)
    }

    fn bio_nearest_ranges(
        &self,
        other: &DataFrame,
        options: BioNearestOptions,
    ) -> Result<DataFrame> {
        if options.direction == BioNearestDirection::Any {
            let (prepared_other, match_by) = prepare_binary_match_by(
                self,
                other,
                options.strand_behavior,
                &options.match_by,
                &options.chromosome_column,
                &options.strand_column,
            )?;

            return self.nearest_ranges(
                prepared_other.as_ref(),
                NearestOptions {
                    match_by,
                    suffix: options.suffix,
                    exclude_overlaps: options.exclude_overlaps,
                    k: options.k,
                    distance_column: options.distance_column,
                    direction: NearestDirection::Any,
                    preserve_input_order: options.preserve_input_order,
                    ..NearestOptions::default()
                },
            );
        }

        let resolved_behavior =
            resolve_strand_behavior(self, other, options.strand_behavior, &options.strand_column)?;

        if resolved_behavior == StrandBehavior::Ignore {
            return Err(RangeFrameError::DirectionalNearestRequiresStrand);
        }

        let row_id_col = unique_row_id_column(self);
        let indexed_self = with_row_index_column(self, &row_id_col)?;
        let (prepared_other, match_by) = prepare_binary_match_by(
            &indexed_self,
            other,
            resolved_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;
        let (forward_self, reverse_self) = split_on_strand(&indexed_self, &options.strand_column)?;

        let forward_direction = match options.direction {
            BioNearestDirection::Downstream => NearestDirection::Forward,
            BioNearestDirection::Upstream => NearestDirection::Backward,
            BioNearestDirection::Any => unreachable!("handled above"),
        };
        let reverse_direction = match options.direction {
            BioNearestDirection::Downstream => NearestDirection::Backward,
            BioNearestDirection::Upstream => NearestDirection::Forward,
            BioNearestDirection::Any => unreachable!("handled above"),
        };

        let mut frames = Vec::new();

        if forward_self.height() > 0 {
            frames.push(forward_self.nearest_ranges(
                prepared_other.as_ref(),
                NearestOptions {
                    match_by: match_by.clone(),
                    suffix: options.suffix.clone(),
                    exclude_overlaps: options.exclude_overlaps,
                    k: options.k,
                    distance_column: options.distance_column.clone(),
                    direction: forward_direction,
                    preserve_input_order: options.preserve_input_order,
                    ..NearestOptions::default()
                },
            )?);
        }

        if reverse_self.height() > 0 {
            frames.push(reverse_self.nearest_ranges(
                prepared_other.as_ref(),
                NearestOptions {
                    match_by,
                    suffix: options.suffix,
                    exclude_overlaps: options.exclude_overlaps,
                    k: options.k,
                    distance_column: options.distance_column,
                    direction: reverse_direction,
                    preserve_input_order: options.preserve_input_order,
                    ..NearestOptions::default()
                },
            )?);
        }

        if frames.is_empty() {
            return drop_column(take_rows(&indexed_self, &[])?, &row_id_col);
        }

        let mut combined = concat_frames(frames)?;
        if options.preserve_input_order {
            combined = sort_rows_by_u32_column(&combined, &row_id_col)?;
        }
        drop_column(combined, &row_id_col)
    }

    fn bio_subtract_overlaps(
        &self,
        other: &DataFrame,
        options: BioSubtractOptions,
    ) -> Result<DataFrame> {
        let (prepared_other, match_by) = prepare_binary_match_by(
            self,
            other,
            options.strand_behavior,
            &options.match_by,
            &options.chromosome_column,
            &options.strand_column,
        )?;

        self.subtract_overlaps(
            prepared_other.as_ref(),
            SubtractOptions {
                match_by,
                preserve_input_order: options.preserve_input_order,
            },
        )
    }

    fn bio_has_valid_strand(&self, strand_column: Option<&str>) -> bool {
        strand_valid(self, strand_column.unwrap_or(DEFAULT_STRAND_COL)).unwrap_or(false)
    }
}

fn prepare_single_match_by(
    df: &DataFrame,
    use_strand: UseStrand,
    match_by: &[String],
    chromosome_column: &str,
    strand_column: &str,
) -> Result<Vec<String>> {
    let mut columns = vec![chromosome_column.to_owned()];

    let include_strand = match use_strand {
        UseStrand::Auto => strand_valid(df, strand_column)?,
        UseStrand::Enabled => {
            if df.column(strand_column).is_err() {
                return Err(RangeFrameError::MissingStrandColumn {
                    column: strand_column.to_owned(),
                });
            }
            true
        }
        UseStrand::Disabled => false,
    };

    if include_strand {
        columns.push(strand_column.to_owned());
    }
    columns.extend(match_by.iter().cloned());

    Ok(dedupe_preserving_order(columns))
}

fn use_strand_enabled(df: &DataFrame, use_strand: UseStrand, strand_column: &str) -> Result<bool> {
    match use_strand {
        UseStrand::Auto => strand_valid(df, strand_column),
        UseStrand::Enabled => {
            if df.column(strand_column).is_err() {
                return Err(RangeFrameError::MissingStrandColumn {
                    column: strand_column.to_owned(),
                });
            }
            Ok(true)
        }
        UseStrand::Disabled => Ok(false),
    }
}

fn prepare_binary_match_by<'a>(
    left: &DataFrame,
    right: &'a DataFrame,
    strand_behavior: StrandBehavior,
    match_by: &[String],
    chromosome_column: &str,
    strand_column: &str,
) -> Result<(Cow<'a, DataFrame>, Vec<String>)> {
    let resolved_behavior = resolve_strand_behavior(left, right, strand_behavior, strand_column)?;

    let mut columns = vec![chromosome_column.to_owned()];
    if resolved_behavior != StrandBehavior::Ignore {
        columns.push(strand_column.to_owned());
    }
    columns.extend(match_by.iter().cloned());
    let match_by = dedupe_preserving_order(columns);

    let prepared_right = if resolved_behavior == StrandBehavior::Opposite {
        Cow::Owned(flip_strand(right, strand_column)?)
    } else {
        Cow::Borrowed(right)
    };

    Ok((prepared_right, match_by))
}

fn resolve_strand_behavior(
    left: &DataFrame,
    right: &DataFrame,
    strand_behavior: StrandBehavior,
    strand_column: &str,
) -> Result<StrandBehavior> {
    match strand_behavior {
        StrandBehavior::Auto => {
            if strand_valid(left, strand_column)? && strand_valid(right, strand_column)? {
                Ok(StrandBehavior::Same)
            } else {
                Ok(StrandBehavior::Ignore)
            }
        }
        StrandBehavior::Same | StrandBehavior::Opposite => {
            if strand_valid(left, strand_column)? && strand_valid(right, strand_column)? {
                Ok(strand_behavior)
            } else {
                Err(RangeFrameError::InvalidStrandBehavior {
                    behavior: strand_behavior.label().to_owned(),
                })
            }
        }
        StrandBehavior::Ignore => Ok(StrandBehavior::Ignore),
    }
}

fn strand_valid(df: &DataFrame, strand_column: &str) -> Result<bool> {
    let column = match df.column(strand_column) {
        Ok(column) => column,
        Err(_) => return Ok(false),
    };

    if column.null_count() > 0 {
        return Ok(false);
    }

    let series = column.as_materialized_series();
    let mut values = match borrowed_string_values(series) {
        Ok(values) => values,
        Err(RangeFrameError::UnsupportedKeyDtype { .. }) => return Ok(false),
        Err(err) => return Err(err),
    };

    Ok(values.all(|value| matches!(value, POSITIVE_STRAND | NEGATIVE_STRAND)))
}

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

    let swapped = borrowed_string_values(column.as_materialized_series())?
        .map(|value| match value {
            POSITIVE_STRAND => NEGATIVE_STRAND,
            NEGATIVE_STRAND => POSITIVE_STRAND,
            other => other,
        })
        .collect::<Vec<_>>();

    let mut out = df.clone();
    out.with_column(Column::from(Series::new(strand_column.into(), swapped)))?;
    Ok(out)
}

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

    let mut plus = Vec::new();
    let mut minus = Vec::new();

    for (row_idx, strand) in borrowed_string_values(column.as_materialized_series())?.enumerate() {
        let row_idx = row_idx as u32;
        match strand {
            POSITIVE_STRAND => plus.push(row_idx),
            NEGATIVE_STRAND => minus.push(row_idx),
            _ => {}
        }
    }

    Ok((take_rows(df, &plus)?, take_rows(df, &minus)?))
}

fn take_non_overlapping_rows(df: &DataFrame, overlapping_rows: Vec<u32>) -> Result<DataFrame> {
    let mut overlapping = vec![false; df.height()];
    for row_id in overlapping_rows {
        if let Some(slot) = overlapping.get_mut(row_id as usize) {
            *slot = true;
        }
    }

    let remaining = overlapping
        .into_iter()
        .enumerate()
        .filter_map(|(row_idx, is_overlapping)| (!is_overlapping).then_some(row_idx as u32))
        .collect::<Vec<_>>();

    take_rows(df, &remaining)
}

fn with_row_index_column(df: &DataFrame, column_name: &str) -> Result<DataFrame> {
    ensure_row_id_capacity(df.height())?;

    let mut out = df.clone();
    let row_ids = (0..df.height() as u32).collect::<Vec<_>>();
    out.with_column(Column::from(Series::new(column_name.into(), row_ids)))?;
    Ok(out)
}

fn unique_row_id_column(df: &DataFrame) -> String {
    if df.column(TEMP_ROW_ID_PREFIX).is_err() {
        return TEMP_ROW_ID_PREFIX.to_owned();
    }

    let mut suffix = 1_usize;
    loop {
        let candidate = format!("{TEMP_ROW_ID_PREFIX}_{suffix}");
        if df.column(&candidate).is_err() {
            return candidate;
        }
        suffix = suffix.saturating_add(1);
    }
}

fn sort_rows_by_u32_column(df: &DataFrame, column_name: &str) -> Result<DataFrame> {
    let column = df.column(column_name)?;
    let order = column
        .as_materialized_series()
        .u32()?
        .into_no_null_iter()
        .enumerate()
        .map(|(position, row_id)| (position as u32, row_id))
        .collect::<Vec<_>>();

    let mut order = order;
    order.sort_unstable_by_key(|(_, row_id)| *row_id);

    let indices = order
        .into_iter()
        .map(|(position, _)| position)
        .collect::<Vec<_>>();

    take_rows_owned(df, indices)
}

fn concat_frames(mut frames: Vec<DataFrame>) -> Result<DataFrame> {
    let mut iter = frames.drain(..);
    let mut combined = iter
        .next()
        .expect("concat_frames is only called with at least one frame");

    for frame in iter {
        combined.vstack_mut(&frame)?;
    }

    Ok(combined)
}

fn drop_column(mut df: DataFrame, column_name: &str) -> Result<DataFrame> {
    let _ = df.drop_in_place(column_name)?;
    Ok(df)
}

fn ensure_row_id_capacity(len: usize) -> Result<()> {
    if len > u32::MAX as usize {
        return Err(RangeFrameError::TooManyRows { len });
    }

    Ok(())
}

fn dedupe_preserving_order(values: Vec<String>) -> Vec<String> {
    let mut out = Vec::with_capacity(values.len());

    for value in values {
        if !out.iter().any(|existing| existing == &value) {
            out.push(value);
        }
    }

    out
}

impl StrandBehavior {
    fn label(self) -> &'static str {
        match self {
            Self::Auto => "auto",
            Self::Same => "same",
            Self::Opposite => "opposite",
            Self::Ignore => "ignore",
        }
    }
}

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

    use super::{
        BioClusterOptions, BioDataFrameRanges, BioMergeOptions, BioNearestDirection,
        BioNearestOptions, BioOverlapOptions, BioSubtractOptions, StrandBehavior, NEGATIVE_STRAND,
        POSITIVE_STRAND,
    };
    use crate::OverlapMode;

    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 bio_merge_defaults_to_chromosome_matching() {
        let df = make_df(vec![
            Series::new("Chromosome".into(), &["chr1", "chr1", "chr2"]),
            Series::new("Start".into(), &[1_i64, 4, 1]),
            Series::new("End".into(), &[5_i64, 8, 2]),
        ]);

        let result = df.bio_merge_overlaps(BioMergeOptions::default()).unwrap();

        assert_eq!(
            result
                .column("Chromosome")
                .unwrap()
                .as_materialized_series()
                .str()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec!["chr1", "chr2"]
        );
        assert_eq!(
            result
                .column("Start")
                .unwrap()
                .as_materialized_series()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![1, 1]
        );
        assert_eq!(
            result
                .column("End")
                .unwrap()
                .as_materialized_series()
                .i64()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec![8, 2]
        );
    }

    #[test]
    fn bio_overlap_supports_invert() {
        let left = make_df(vec![
            Series::new("Chromosome".into(), &["chr1", "chr1", "chr2"]),
            Series::new("Start".into(), &[1_i64, 1, 10]),
            Series::new("End".into(), &[3_i64, 3, 11]),
            Series::new("ID".into(), &["A", "a", "b"]),
        ]);
        let right = make_df(vec![
            Series::new("Chromosome".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[2_i64, 2]),
            Series::new("End".into(), &[3_i64, 9]),
        ]);

        let multiple = left
            .bio_overlap_ranges(
                &right,
                BioOverlapOptions {
                    multiple: OverlapMode::All,
                    ..BioOverlapOptions::default()
                },
            )
            .unwrap();
        let inverted = left
            .bio_overlap_ranges(
                &right,
                BioOverlapOptions {
                    invert: true,
                    ..BioOverlapOptions::default()
                },
            )
            .unwrap();

        assert_eq!(
            multiple
                .column("ID")
                .unwrap()
                .as_materialized_series()
                .str()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec!["A", "A", "a", "a"]
        );
        assert_eq!(
            inverted
                .column("ID")
                .unwrap()
                .as_materialized_series()
                .str()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec!["b"]
        );
    }

    #[test]
    fn bio_nearest_supports_downstream_direction() {
        let left = make_df(vec![
            Series::new("Chromosome".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[10_i64, 10]),
            Series::new("End".into(), &[12_i64, 12]),
            Series::new("Strand".into(), &[POSITIVE_STRAND, NEGATIVE_STRAND]),
            Series::new("Name".into(), &["plus", "minus"]),
        ]);
        let right = make_df(vec![
            Series::new("Chromosome".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[20_i64, 1]),
            Series::new("End".into(), &[21_i64, 2]),
            Series::new("Strand".into(), &[POSITIVE_STRAND, NEGATIVE_STRAND]),
            Series::new("Hit".into(), &["plus_hit", "minus_hit"]),
        ]);

        let result = left
            .bio_nearest_ranges(
                &right,
                BioNearestOptions {
                    direction: BioNearestDirection::Downstream,
                    ..BioNearestOptions::default()
                },
            )
            .unwrap();

        let sorted = result.sort(["Name"], Default::default()).unwrap();
        assert_eq!(
            sorted
                .column("Hit_b")
                .unwrap()
                .as_materialized_series()
                .str()
                .unwrap()
                .into_no_null_iter()
                .collect::<Vec<_>>(),
            vec!["minus_hit", "plus_hit"]
        );
    }

    #[test]
    fn bio_cluster_uses_strand_when_requested() {
        let df = make_df(vec![
            Series::new("Chromosome".into(), &["chr1", "chr1"]),
            Series::new("Start".into(), &[1_i64, 1]),
            Series::new("End".into(), &[3_i64, 3]),
            Series::new("Strand".into(), &[POSITIVE_STRAND, NEGATIVE_STRAND]),
        ]);

        let result = df
            .bio_cluster_overlaps(BioClusterOptions {
                use_strand: super::UseStrand::Enabled,
                ..BioClusterOptions::default()
            })
            .unwrap();

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

    #[test]
    fn bio_subtract_supports_opposite_strand_behavior() {
        let left = make_df(vec![
            Series::new("Chromosome".into(), &["chr1"]),
            Series::new("Start".into(), &[1_i64]),
            Series::new("End".into(), &[10_i64]),
            Series::new("Strand".into(), &[POSITIVE_STRAND]),
        ]);
        let right = make_df(vec![
            Series::new("Chromosome".into(), &["chr1"]),
            Series::new("Start".into(), &[4_i64]),
            Series::new("End".into(), &[6_i64]),
            Series::new("Strand".into(), &[NEGATIVE_STRAND]),
        ]);

        let result = left
            .bio_subtract_overlaps(
                &right,
                BioSubtractOptions {
                    strand_behavior: StrandBehavior::Opposite,
                    ..BioSubtractOptions::default()
                },
            )
            .unwrap();

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