joinp.rs

static USAGE: &str = r#"
Joins two sets of CSV data on the specified columns using the Polars engine.

The default join operation is an 'inner' join. This corresponds to the
intersection of rows on the keys specified.

Unlike the join command, joinp can process files larger than RAM, is multithreaded,
has join key validation, a maintain row order option, pre-join filtering, supports
asof joins and its output columns can be coalesced (no duplicate columns).

Returns the shape of the join result (number of rows, number of columns) to stderr.

For examples, see https://github.com/dathere/qsv/blob/master/tests/test_joinp.rs.

Usage:
    qsv joinp [options] <columns1> <input1> <columns2> <input2>
    qsv joinp --cross [--validate <arg>] <input1> <input2> [--output <file>]
    qsv joinp --non-equi <expr> <input1> <input2> [options] [--output <file>]
    qsv joinp --help

joinp arguments:
    Both <input1> aka <left> & <input2> aka <right> files need to have headers.
    Stdin is not supported.

    The columns arguments specify the columns to join for each input. Columns are
    referenced by name. Specify multiple columns by separating them with a comma.
    Both <columns1> and <columns2> must specify exactly the same number of columns.

    Note that <input1> is the left CSV data set and <input2> is the right CSV data set.

joinp options:
    -i, --ignore-case      When set, joins are done case insensitively.
-z, --ignore-leading-zeros  When set, joins are done ignoring leading zeros.
                           Note that this is only applied to the join keys for
                           both numeric and string columns. The output columns
                           will not have leading zeros.
    --left                 Do a 'left outer' join. This returns all rows in
                           first CSV data set, including rows with no
                           corresponding row in the second data set. When no
                           corresponding row exists, it is padded out with
                           empty fields.
    --left-anti            This returns only the rows in the first CSV data set
                           that do not have a corresponding row in the second
                           data set. The output schema is the same as the
                           first dataset.
    --left-semi            This returns only the rows in the first CSV data set
                           that have a corresponding row in the second data set.
                           The output schema is the same as the first data set.
    --right                Do a 'right outer' join. This returns all rows in
                           second CSV data set, including rows with no
                           corresponding row in the first data set. When no
                           corresponding row exists, it is padded out with
                           empty fields. (This is the reverse of 'outer left'.)
    --right-anti           This returns only the rows in the second CSV data set
                           that do not have a corresponding row in the first
                           data set. The output schema is the same as the
                           second dataset.
    --right-semi           This returns only the rows in the second CSV data set
                           that have a corresponding row in the first data set.
                           The output schema is the same as the second data set.
    --full                 Do a 'full outer' join. This returns all rows in
                           both data sets with matching records joined. If
                           there is no match, the missing side will be padded
                           out with empty fields.
    --cross                USE WITH CAUTION.
                           This returns the cartesian product of the CSV
                           data sets given. The number of rows return is
                           equal to N * M, where N and M correspond to the
                           number of rows in the given data sets, respectively.
                           The columns1 and columns2 arguments are ignored.
    --non-equi <expr>      Do a non-equi join. The given expression is evaluated
                           for each row in the left dataset and can refer to columns
                           in the left and right dataset. If the expression evaluates
                           to true, the row is joined with the corresponding row in
                           the right dataset.
                           The expression is a valid Polars SQL where clause, with each
                           column name followed by "_left" or "_right" suffixes to indicate
                           which data set the column belongs to.
                           (e.g. "salary_left >= min_salary_right AND \
                                  salary_left <= max_salary_right AND \
                                  experience_left >= min_exp_right")

    --coalesce             Force the join to coalesce columns with the same name.
                           For inner joins, this is not necessary as the join
                           columns are automatically coalesced.

    --filter-left <arg>    Filter the left CSV data set by the given Polars SQL
                           expression BEFORE the join. Only rows that evaluates
                           to true are used in the join.
    --filter-right <arg>   Filter the right CSV data set by the given Polars SQL
                           expression BEFORE the join. Only rows that evaluates
                           to true are used in the join.
    --validate <arg>       Validate the join keys BEFORE performing the join.
                           Valid values are:
                             none - No validation is performed.
                             onetomany - join keys are unique in the left data set.
                             manytoone - join keys are unique in the right data set.
                             onetoone - join keys are unique in both left & right data sets.
                           [default: none]

                            JOIN OPTIONS:
    --maintain-order <arg>  Which row order to preserve, if any. Valid values are:
                              none, left, right, left_right, right_left
                            Do not rely on any observed ordering without explicitly
                            setting this parameter. Not specifying any order can improve
                            performance. Supported for inner, left, right and full joins.
                            [default: none]
    --nulls                When set, joins will work on empty fields.
                           Otherwise, empty fields are completely ignored.
    --streaming            When set, the join will be done in a streaming fashion.
                           Only use this when you get out of memory errors.

                           POLARS CSV PARSING OPTIONS:
    --try-parsedates       When set, will attempt to parse the columns as dates.
                           If the parse fails, columns remain as strings.
                           This is useful when the join keys are formatted as 
                           dates with differing date formats, as the date formats
                           will be normalized. Note that this will be automatically 
                           enabled when using asof joins.
    --infer-len <arg>      The number of rows to scan when inferring the schema of the CSV.
                           Set to 0 to do a full table scan (warning: very slow).
                           Only used when --cache-schema is 0 or 1 and no cached schema exists or
                           when --infer-len is 0.
                           [default: 10000]
    --cache-schema <arg>   Create and cache Polars schema JSON files.
                           Ignored when --infer-len is 0.
                           ‎ -2: treat all columns as String. A Polars schema file is created & cached.
                           ‎ -1: treat all columns as String. No Polars schema file is created.
                             0: do not cache Polars schema. Uses --infer-len to infer schema.
                             1: cache Polars schema with the following behavior:
                                - If schema file exists and is newer than input: use cached schema
                                - If schema file missing/outdated and stats cache exists: 
                                  derive schema from stats and cache it
                                - If no schema or stats cache: infer schema using --infer-len 
                                  and cache the result
                                Schema files use the same name as input with .pschema.json extension
                                (e.g., data.csv -> data.pschema.json)
                           [default: 0]
    --low-memory           Use low memory mode when parsing CSVs. This will use less memory
                           but will be slower. It will also process the join in streaming mode.
                           Only use this when you get out of memory errors.
    --no-optimizations     Disable non-default join optimizations. This will make joins slower.
                           Only use this when you get join errors.                           
    --ignore-errors        Ignore errors when parsing CSVs. If set, rows with errors
                           will be skipped. If not set, the query will fail.
                           Only use this when debugging queries, as polars does batched
                           parsing and will skip the entire batch where the error occurred.
                           To get more detailed error messages, set the environment variable
                           POLARS_BACKTRACE_IN_ERR=1 before running the join.
    --decimal-comma        Use comma as the decimal separator when parsing CSVs.
                           Otherwise, use period as the decimal separator.
                           Note that you'll need to set --delimiter to an alternate delimiter
                           other than the default comma if you are using this option.

                           ASOF JOIN OPTIONS:
    --asof                 Do an 'asof' join. This is similar to a left inner
                           join, except we match on nearest key rather than
                           equal keys. Note that both CSV data sets will be SORTED
                           AUTOMATICALLY on the join columns.
                           Particularly useful for time series data.
    --left_by <arg>        Do an 'asof_by' join - a special implementation of the asof
                           join that searches for the nearest keys within a subgroup
                           set by the asof_by columns. This specifies the column/s for
                           the left CSV. Columns are referenced by name. Specify
                           multiple columns by separating them with a comma.
    --right_by <arg>       Do an 'asof_by' join. This specifies the column/s for
                           the right CSV.     
    --strategy <arg>       The strategy to use for the asof join:
                             backward - For each row in the first CSV data set,
                                        we find the last row in the second data set
                                        whose key is less than or equal to the key
                                        in the first data set.
                             forward -  For each row in the first CSV data set,
                                        we find the first row in the second data set
                                        whose key is greater than or equal to the key
                                        in the first data set.
                             nearest -  selects the last row in the second data set
                                        whose value is nearest to the value in the
                                        first data set.
                           [default: backward]
    --tolerance <arg>      The tolerance for the nearest asof join. This is only
                           used when the nearest strategy is used. The
                           tolerance is a positive integer that specifies
                           the maximum number of rows to search for a match.

                           If the join is done on a column of type Date, Time or
                           DateTime, then the tolerance is interpreted using
                           the following language:
                                1d - 1 day
                                1h - 1 hour
                                1m - 1 minute
                                1s - 1 second
                                1ms - 1 millisecond
                                1us - 1 microsecond
                                1ns - 1 nanosecond
                                1w - 1 week
                                1mo - 1 month
                                1q - 1 quarter
                                1y - 1 year
                                1i - 1 index count
                             Or combine them: “3d12h4m25s” # 3 days, 12 hours,
                             4 minutes, and 25 seconds
                             Suffix with “_saturating” to indicate that dates too
                             large for their month should saturate at the largest date
                             (e.g. 2022-02-29 -> 2022-02-28) instead of erroring.

                             OUTPUT FORMAT OPTIONS:
   --sql-filter <SQL>        The SQL expression to apply against the join result.
                             Used to select columns and filter rows AFTER running the join.
                             Be sure to select from the "join_result" table when formulating
                             the SQL expression.
                             (e.g. "select c1, c2 as colname from join_result where c2 > 20")
   --datetime-format <fmt>   The datetime format to use writing datetimes.
                             See https://docs.rs/chrono/latest/chrono/format/strftime/index.html
                             for the list of valid format specifiers.
   --date-format <fmt>       The date format to use writing dates.
   --time-format <fmt>       The time format to use writing times.
   --float-precision <arg>   The number of digits of precision to use when writing floats.
                             (default: 6)
   --null-value <arg>        The string to use when writing null values.
                             (default: <empty string>)

Common options:
    -h, --help             Display this message
    -o, --output <file>    Write output to <file> instead of stdout.
    -d, --delimiter <arg>  The field delimiter for reading/writing CSV data.
                           Must be a single character. (default: ,)
    -Q, --quiet            Do not return join shape to stderr.
"#;

use std::{
    env,
    fs::File,
    io::{self, BufReader, BufWriter, Read, Write},
    mem::swap,
    path::{Path, PathBuf},
    str,
};

use polars::{datatypes::AnyValue, prelude as pl, prelude::*, sql::SQLContext};
use serde::Deserialize;
use tempfile::tempdir;

use crate::{
    cmd::sqlp::compress_output_if_needed, config::Delimiter, util, util::get_stats_records,
    CliResult,
};

#[derive(Deserialize)]
struct Args {
    arg_columns1:              String,
    arg_input1:                String,
    arg_columns2:              String,
    arg_input2:                String,
    flag_left:                 bool,
    flag_left_anti:            bool,
    flag_left_semi:            bool,
    flag_right:                bool,
    flag_right_anti:           bool,
    flag_right_semi:           bool,
    flag_full:                 bool,
    flag_cross:                bool,
    flag_non_equi:             Option<String>,
    flag_coalesce:             bool,
    flag_filter_left:          Option<String>,
    flag_filter_right:         Option<String>,
    flag_validate:             Option<String>,
    flag_maintain_order:       Option<String>,
    flag_nulls:                bool,
    flag_streaming:            bool,
    flag_try_parsedates:       bool,
    flag_decimal_comma:        bool,
    flag_infer_len:            usize,
    flag_cache_schema:         i8,
    flag_low_memory:           bool,
    flag_no_optimizations:     bool,
    flag_ignore_errors:        bool,
    flag_asof:                 bool,
    flag_left_by:              Option<String>,
    flag_right_by:             Option<String>,
    flag_strategy:             Option<String>,
    flag_tolerance:            Option<String>,
    flag_sql_filter:           Option<String>,
    flag_datetime_format:      Option<String>,
    flag_date_format:          Option<String>,
    flag_time_format:          Option<String>,
    flag_float_precision:      Option<usize>,
    flag_null_value:           String,
    flag_output:               Option<String>,
    flag_delimiter:            Option<Delimiter>,
    flag_quiet:                bool,
    flag_ignore_case:          bool,
    flag_ignore_leading_zeros: bool,
}

#[derive(PartialEq, Eq)]
enum SpecialJoin {
    NonEqui(String),
    AsOf,
    None,
}

pub fn run(argv: &[&str]) -> CliResult<()> {
    let mut args: Args = util::get_args(USAGE, argv)?;

    // always try to parse dates when its an asof join
    // just in case the user doesn't specify it
    // and they're using date/time/datetime columns
    if args.flag_asof {
        args.flag_try_parsedates = true;
    }

    let tmpdir = tempdir()?;
    let join = args.new_join(&tmpdir)?;

    let flag_validate = args
        .flag_validate
        .unwrap_or_else(|| "none".to_string())
        .to_lowercase();
    let validation = match flag_validate.as_str() {
        // no unique checks
        "manytomany" | "none" => JoinValidation::ManyToMany,
        // join keys are unique in the left data set
        "onetomany" => JoinValidation::OneToMany,
        // join keys are unique in the right data set
        "manytoone" => JoinValidation::ManyToOne,
        // join keys are unique in both left & right data sets
        "onetoone" => JoinValidation::OneToOne,
        s => return fail_incorrectusage_clierror!("Invalid join validation: {s}"),
    };

    let flag_maintain_order = args
        .flag_maintain_order
        .unwrap_or_else(|| "none".to_string())
        .to_lowercase();
    let maintain_order = match flag_maintain_order.as_str() {
        "none" => MaintainOrderJoin::None,
        "left" => MaintainOrderJoin::Left,
        "right" => MaintainOrderJoin::Right,
        "left_right" => MaintainOrderJoin::LeftRight,
        "right_left" => MaintainOrderJoin::RightLeft,
        s => return fail_incorrectusage_clierror!("Invalid maintain order option: {s}"),
    };

    let join_shape: (usize, usize) = match (
        args.flag_left,
        args.flag_left_anti,
        args.flag_left_semi,
        args.flag_right,
        args.flag_right_anti,
        args.flag_right_semi,
        args.flag_full,
        args.flag_cross,
        args.flag_asof,
        args.flag_non_equi.is_some(),
    ) {
        // default inner join
        (false, false, false, false, false, false, false, false, false, false) => join.run(
            JoinType::Inner,
            validation,
            maintain_order,
            SpecialJoin::None,
        ),
        // left join
        (true, false, false, false, false, false, false, false, false, false) => join.run(
            JoinType::Left,
            validation,
            maintain_order,
            SpecialJoin::None,
        ),
        // left anti join
        (false, true, false, false, false, false, false, false, false, false) => join.run(
            JoinType::Anti,
            validation,
            maintain_order,
            SpecialJoin::None,
        ),
        // left semi join
        (false, false, true, false, false, false, false, false, false, false) => join.run(
            JoinType::Semi,
            validation,
            maintain_order,
            SpecialJoin::None,
        ),
        // right join
        (false, false, false, true, false, false, false, false, false, false) => join.run(
            JoinType::Right,
            validation,
            maintain_order,
            SpecialJoin::None,
        ),
        // right anti join
        // swap left and right data sets and run left anti join
        (false, false, false, false, true, false, false, false, false, false) => {
            let mut swapped_join = join;
            swap(&mut swapped_join.left_lf, &mut swapped_join.right_lf);
            swap(&mut swapped_join.left_sel, &mut swapped_join.right_sel);
            swapped_join.run(
                JoinType::Anti,
                validation,
                maintain_order,
                SpecialJoin::None,
            )
        },
        // right semi join
        // swap left and right data sets and run left semi join
        (false, false, false, false, false, true, false, false, false, false) => {
            let mut swapped_join = join;
            swap(&mut swapped_join.left_lf, &mut swapped_join.right_lf);
            swap(&mut swapped_join.left_sel, &mut swapped_join.right_sel);
            swapped_join.run(
                JoinType::Semi,
                validation,
                maintain_order,
                SpecialJoin::None,
            )
        },
        // full join
        (false, false, false, false, false, false, true, false, false, false) => join.run(
            JoinType::Full,
            validation,
            maintain_order,
            SpecialJoin::None,
        ),
        // cross join
        (false, false, false, false, false, false, false, true, false, false) => join.run(
            JoinType::Cross,
            validation,
            MaintainOrderJoin::None,
            SpecialJoin::None,
        ),

        // as of join
        (false, false, false, false, false, false, false, false, true, false) => {
            // safety: flag_strategy is always is_some() as it has a default value
            args.flag_strategy = Some(args.flag_strategy.unwrap().to_lowercase());
            let strategy = match args.flag_strategy.as_deref() {
                Some("backward") | None => AsofStrategy::Backward,
                Some("forward") => AsofStrategy::Forward,
                Some("nearest") => AsofStrategy::Nearest,
                Some(s) => return fail_incorrectusage_clierror!("Invalid asof strategy: {}", s),
            };

            let mut asof_options = AsOfOptions {
                strategy,
                ..Default::default()
            };

            if strategy == AsofStrategy::Nearest {
                if let Some(ref tolerance) = args.flag_tolerance {
                    // If the tolerance is a positive integer, it is tolerance number of rows.
                    // Otherwise, it is a tolerance date language spec.
                    if let Ok(numeric_tolerance) = atoi_simd::parse_pos::<u64>(tolerance.as_bytes())
                    {
                        asof_options.tolerance = Some(AnyValue::UInt64(numeric_tolerance));
                    } else {
                        asof_options.tolerance_str = Some(tolerance.into());
                    }
                }
            }
            if args.flag_left_by.is_some() {
                asof_options.left_by = Some(
                    args.flag_left_by
                        .unwrap()
                        .split(',')
                        .map(PlSmallStr::from_str)
                        .collect(),
                );
            }
            if args.flag_right_by.is_some() {
                asof_options.right_by = Some(
                    args.flag_right_by
                        .unwrap()
                        .split(',')
                        .map(PlSmallStr::from_str)
                        .collect(),
                );
            }
            join.run(
                JoinType::AsOf(asof_options),
                validation,
                MaintainOrderJoin::None,
                SpecialJoin::AsOf,
            )
        },

        // non-equi join
        (false, false, false, false, false, false, false, false, false, true) => {
            // JoinType::Inner is just a placeholder value to satisfy the compiler
            // as this is a non-equi join
            join.run(
                JoinType::Inner,
                validation,
                maintain_order,
                SpecialJoin::NonEqui(args.flag_non_equi.unwrap()),
            )
        },
        _ => fail_incorrectusage_clierror!("Please pick exactly one join operation."),
    }?;

    if !args.flag_quiet {
        eprintln!("{join_shape:?}");
    }

    Ok(())
}

struct JoinStruct {
    left_lf:              LazyFrame,
    left_sel:             String,
    right_lf:             LazyFrame,
    right_sel:            String,
    output:               Option<String>,
    delim:                u8,
    coalesce:             bool,
    streaming:            bool,
    no_optimizations:     bool,
    sql_filter:           Option<String>,
    datetime_format:      Option<String>,
    date_format:          Option<String>,
    time_format:          Option<String>,
    float_precision:      Option<usize>,
    null_value:           String,
    ignore_case:          bool,
    ignore_leading_zeros: bool,
}

impl JoinStruct {
    fn run(
        mut self,
        jointype: JoinType,
        validation: JoinValidation,
        maintain_order: MaintainOrderJoin,
        special_join: SpecialJoin,
    ) -> CliResult<(usize, usize)> {
        let mut left_selcols: Vec<_> = self
            .left_sel
            .split(',')
            .map(polars::lazy::dsl::col)
            .collect();
        let mut right_selcols: Vec<_> = self
            .right_sel
            .split(',')
            .map(polars::lazy::dsl::col)
            .collect();

        // If ignore_case is enabled, create lowercase versions of the join columns
        // but only if ignore_leading_zeros is not enabled
        // because leading zeros handling does its own lowercase conversion
        if self.ignore_case && !self.ignore_leading_zeros {
            // Create temporary lowercase versions of join columns in left dataframe
            for col in &left_selcols {
                self.left_lf = self
                    .left_lf
                    .with_column(col.clone().str().to_lowercase().alias(format!(
                        "_qsv-{}-lower",
                        col.to_string()
                            .trim_start_matches(r#"col(""#)
                            .trim_end_matches(r#"")"#)
                    )));
            }

            // Create temporary lowercase versions of join columns in right dataframe
            for col in &right_selcols {
                self.right_lf = self
                    .right_lf
                    .with_column(col.clone().str().to_lowercase().alias(format!(
                        "_qsv-{}-lower",
                        col.to_string()
                            .trim_start_matches(r#"col(""#)
                            .trim_end_matches(r#"")"#)
                    )));
            }

            // Create new vectors for the lowercase column names
            let left_selcols_w: Vec<_> = left_selcols
                .iter()
                .map(|col| {
                    polars::lazy::dsl::col(format!(
                        "_qsv-{}-lower",
                        col.to_string()
                            .trim_start_matches(r#"col(""#)
                            .trim_end_matches(r#"")"#)
                    ))
                })
                .collect();
            left_selcols = left_selcols_w;
            let right_selcols_w: Vec<_> = right_selcols
                .iter()
                .map(|col| {
                    polars::lazy::dsl::col(format!(
                        "_qsv-{}-lower",
                        col.to_string()
                            .trim_start_matches(r#"col(""#)
                            .trim_end_matches(r#"")"#)
                    ))
                })
                .collect();
            right_selcols = right_selcols_w;
        }

        let left_selcols_len = left_selcols.len();
        let right_selcols_len = right_selcols.len();

        if left_selcols_len != right_selcols_len {
            return fail_incorrectusage_clierror!(
                "Both columns1 ({left_selcols:?}) and columns2 ({right_selcols:?}) must specify \
                 the same number of columns ({left_selcols_len } != {right_selcols_len})."
            );
        }

        let coalesce_flag = if self.coalesce {
            JoinCoalesce::CoalesceColumns
        } else {
            JoinCoalesce::JoinSpecific
        };

        let mut optflags = OptFlags::from_bits_truncate(0);
        if self.no_optimizations {
            optflags |= OptFlags::TYPE_COERCION;
        } else {
            optflags |= OptFlags::PROJECTION_PUSHDOWN
                | OptFlags::PREDICATE_PUSHDOWN
                | OptFlags::CLUSTER_WITH_COLUMNS
                | OptFlags::TYPE_COERCION
                | OptFlags::SIMPLIFY_EXPR
                | OptFlags::FILE_CACHING
                | OptFlags::SLICE_PUSHDOWN
                | OptFlags::COMM_SUBPLAN_ELIM
                | OptFlags::COMM_SUBEXPR_ELIM
                | OptFlags::ROW_ESTIMATE
                | OptFlags::FAST_PROJECTION;
        }

        optflags.set(OptFlags::STREAMING, self.streaming);

        // log::debug!("Optimization flags: {optimization_flags:?}");

        let join_results = if jointype == JoinType::Cross {
            // cross join doesn't need join columns
            self.left_lf
                .with_optimizations(optflags)
                .join_builder()
                .with(self.right_lf.with_optimizations(optflags))
                .how(JoinType::Cross)
                .coalesce(coalesce_flag)
                .allow_parallel(true)
                .validate(validation)
                .finish()
                .collect()?
        } else {
            if special_join == SpecialJoin::AsOf {
                // it's an asof join
                // sort by the asof columns, as asof joins require sorted join column data
                let left_selcols_vec: Vec<PlSmallStr> =
                    self.left_sel.split(',').map(PlSmallStr::from_str).collect();

                self.left_lf = self
                    .left_lf
                    .sort(left_selcols_vec, SortMultipleOptions::default());

                let right_selcols_vec: Vec<PlSmallStr> = self
                    .right_sel
                    .split(',')
                    .map(PlSmallStr::from_str)
                    .collect();

                self.right_lf = self
                    .right_lf
                    .sort(right_selcols_vec, SortMultipleOptions::default());
            }

            if let SpecialJoin::NonEqui(expr) = special_join {
                // it's a non-equi join
                let expr = polars::sql::sql_expr(expr)?;

                // Add "_left" & "_right" suffixes to all columns before doing the non-equi join.
                // This is necessary as the NonEqui expression is a SQL where clause and the
                // column names for the left and right data sets are used in the expression.
                self.left_lf = self.left_lf.select([all().name().suffix("_left")]);
                self.right_lf = self.right_lf.select([all().name().suffix("_right")]);

                self.left_lf
                    .with_optimizations(optflags)
                    .join_builder()
                    .with(self.right_lf.with_optimizations(optflags))
                    .join_where(vec![expr])
                    .collect()?
            } else {
                // it's one of the "standard" joins as indicated by jointype
                self.left_lf
                    .with_optimizations(optflags)
                    .join_builder()
                    .with(self.right_lf.with_optimizations(optflags))
                    .left_on(left_selcols)
                    .right_on(right_selcols)
                    .how(jointype)
                    .maintain_order(maintain_order)
                    .coalesce(coalesce_flag)
                    .allow_parallel(true)
                    .validate(validation)
                    .finish()
                    .collect()?
            }
        };

        let mut results_df = if let Some(sql_filter) = &self.sql_filter {
            let mut ctx = SQLContext::new();
            ctx.register("join_result", join_results.lazy());
            ctx.execute(sql_filter)
                .and_then(polars::prelude::LazyFrame::collect)?
        } else {
            join_results
        };

        // if self.ignore_case, remove the temporary lowercase columns from the dataframe
        if self.ignore_case && !self.ignore_leading_zeros {
            // Get all column names
            let cols = results_df.get_column_names();
            // Filter out the lowercase columns (those with "_qsv-*-lower" pattern)
            let keep_cols: Vec<String> = cols
                .iter()
                .filter(|&col| !(col.starts_with("_qsv-") && col.ends_with("-lower")))
                .map(|&s| s.to_string())
                .collect();
            // Select only the non-lowercase columns
            results_df = results_df.select(keep_cols)?;
        }

        let mut out_delim = self.delim;
        let mut out_writer = match self.output {
            Some(ref output_file) => {
                out_delim = tsvssv_delim(output_file, self.delim);

                // no need to use buffered writer here, as CsvWriter already does that
                let path = Path::new(&output_file);
                Box::new(File::create(path).unwrap()) as Box<dyn Write>
            },
            None => Box::new(io::stdout()) as Box<dyn Write>,
        };

        // shape is the number of rows and columns
        let join_shape = results_df.shape();

        CsvWriter::new(&mut out_writer)
            .include_header(true)
            .with_separator(out_delim)
            .with_datetime_format(self.datetime_format)
            .with_date_format(self.date_format)
            .with_time_format(self.time_format)
            .with_float_precision(self.float_precision)
            .with_null_value(self.null_value)
            .include_bom(util::get_envvar_flag("QSV_OUTPUT_BOM"))
            .finish(&mut results_df)?;

        compress_output_if_needed(self.output)?;

        Ok(join_shape)
    }
}

impl Args {
    fn new_join(&mut self, tmpdir: &tempfile::TempDir) -> CliResult<JoinStruct> {
        // Helper function to create a LazyFrameReader with common settings
        fn create_lazy_reader(
            file_path: &str,
            comment_char: Option<&PlSmallStr>,
            args: &Args,
            delim: u8,
        ) -> LazyCsvReader {
            LazyCsvReader::new(file_path)
                .with_has_header(true)
                .with_missing_is_null(args.flag_nulls)
                .with_comment_prefix(comment_char.cloned())
                .with_separator(tsvssv_delim(file_path, delim))
                .with_try_parse_dates(args.flag_try_parsedates)
                .with_decimal_comma(args.flag_decimal_comma)
                .with_low_memory(args.flag_low_memory)
                .with_ignore_errors(args.flag_ignore_errors)
        }

        // Helper function to handle schema creation from stats
        fn create_schema_from_stats(input_path: &Path, args: &Args) -> CliResult<Schema> {
            let schema_args = util::SchemaArgs {
                flag_enum_threshold:  0,
                flag_ignore_case:     false,
                flag_strict_dates:    false,
                flag_pattern_columns: crate::select::SelectColumns::parse("").unwrap(),
                flag_dates_whitelist: String::new(),
                flag_prefer_dmy:      false,
                flag_force:           false,
                flag_stdout:          false,
                flag_jobs:            Some(util::njobs(None)),
                flag_no_headers:      false,
                flag_delimiter:       args.flag_delimiter,
                arg_input:            Some(input_path.to_string_lossy().into_owned()),
                flag_memcheck:        false,
            };

            let (csv_fields, csv_stats) =
                get_stats_records(&schema_args, util::StatsMode::PolarsSchema)?;

            let mut schema = Schema::with_capacity(csv_stats.len());
            for (idx, stat) in csv_stats.iter().enumerate() {
                schema.insert(
                    PlSmallStr::from_str(
                        simdutf8::basic::from_utf8(csv_fields.get(idx).unwrap()).unwrap(),
                    ),
                    {
                        let datatype = &stat.r#type;
                        #[allow(clippy::match_same_arms)]
                        match datatype.as_str() {
                            "String" => polars::datatypes::DataType::String,
                            "Integer" => {
                                let min = stat.min.as_ref().unwrap();
                                let max = stat.max.as_ref().unwrap();
                                if min.parse::<i32>().is_ok() && max.parse::<i32>().is_ok() {
                                    polars::datatypes::DataType::Int32
                                } else {
                                    polars::datatypes::DataType::Int64
                                }
                            },
                            "Float" => {
                                let min = stat.min.as_ref().unwrap();
                                let max = stat.max.as_ref().unwrap();
                                if min.parse::<f32>().is_ok() && max.parse::<f32>().is_ok() {
                                    polars::datatypes::DataType::Float32
                                } else {
                                    polars::datatypes::DataType::Float64
                                }
                            },
                            "Boolean" => polars::datatypes::DataType::Boolean,
                            "Date" => polars::datatypes::DataType::Date,
                            _ => polars::datatypes::DataType::String,
                        }
                    },
                );
            }
            Ok(schema)
        }

        /// Helper function to setup a LazyFrame with schema handling based on cache_schema flag.
        ///
        /// # Arguments
        /// * `input_path` - Path to the input CSV file
        /// * `comment_char` - Optional comment character to ignore lines starting with it
        /// * `args` - Command line arguments containing schema caching and other options
        /// * `delim` - Delimiter character for CSV parsing
        /// * `debuglog_flag` - Whether debug logging is enabled
        ///
        /// # Returns
        /// Returns a tuple containing:
        /// * The configured LazyFrame for reading the CSV
        /// * A boolean indicating if a new schema needs to be created and cached
        ///
        /// # Schema Caching Modes
        /// * `0` - No schema caching, infer schema from data sample using Polars
        /// * `1` - Cache inferred schema from stats in .pschema.json file
        /// * `-1` - Use string schema for all columns without caching
        /// * `-2` - Use string schema for all columns and cache it
        ///
        /// # Errors
        /// Returns error if:
        /// * File operations fail
        /// * Schema parsing fails
        /// * Invalid cache_schema value provided
        fn setup_lazy_frame(
            input_path: &Path,
            comment_char: Option<&PlSmallStr>,
            args: &Args,
            delim: u8,
            debuglog_flag: bool,
        ) -> CliResult<(LazyFrame, bool)> {
            let schema_file = input_path.canonicalize()?.with_extension("pschema.json");
            let mut create_schema = false;
            let cache_schema = if args.flag_infer_len == 0 {
                0
            } else {
                args.flag_cache_schema
            };

            let mut reader =
                create_lazy_reader(input_path.to_str().unwrap(), comment_char, args, delim);

            match cache_schema {
                0 => {
                    reader = reader.with_infer_schema_length(if args.flag_infer_len == 0 {
                        None
                    } else {
                        Some(args.flag_infer_len)
                    });
                },
                1 => {
                    let mut valid_schema_exists = schema_file.exists()
                        && schema_file.metadata()?.modified()?
                            > input_path.metadata()?.modified()?;

                    if !valid_schema_exists {
                        let schema = create_schema_from_stats(input_path, args)?;
                        let stats_schema = Arc::new(schema);
                        let stats_schema_json = serde_json::to_string_pretty(&stats_schema)?;

                        let mut file = BufWriter::new(File::create(&schema_file)?);
                        file.write_all(stats_schema_json.as_bytes())?;
                        file.flush()?;
                        if debuglog_flag {
                            log::debug!("Saved schema to file: {}", schema_file.display());
                        }
                        valid_schema_exists = true;
                    }

                    if valid_schema_exists {
                        let file = File::open(&schema_file)?;
                        let mut buf_reader = BufReader::new(file);
                        let mut schema_json = String::with_capacity(100);
                        buf_reader.read_to_string(&mut schema_json)?;
                        let schema: Schema = serde_json::from_str(&schema_json)?;
                        reader = reader.with_schema(Some(Arc::new(schema)));
                        create_schema = false;
                    } else {
                        reader = reader.with_infer_schema_length(Some(args.flag_infer_len));
                        create_schema = true;
                    }
                },
                -1 | -2 => {
                    // get the headers from the input file
                    let mut rdr = csv::Reader::from_path(input_path)?;
                    let csv_fields = rdr.byte_headers()?.clone();
                    drop(rdr);

                    let mut schema = Schema::with_capacity(csv_fields.len());
                    for field in &csv_fields {
                        schema.insert(
                            PlSmallStr::from_str(simdutf8::basic::from_utf8(field).unwrap()),
                            polars::datatypes::DataType::String,
                        );
                    }
                    let allstring_schema = Arc::new(schema);

                    reader = reader.with_schema(Some(allstring_schema.clone()));
                    create_schema = false;

                    // create and cache allstring schema
                    if cache_schema == -2 {
                        let allstring_schema_json =
                            serde_json::to_string_pretty(&allstring_schema)?;

                        let mut file = BufWriter::new(File::create(&schema_file)?);
                        file.write_all(allstring_schema_json.as_bytes())?;
                        file.flush()?;
                        if debuglog_flag {
                            log::debug!(
                                "Saved allstring_schema to file: {}",
                                schema_file.display()
                            );
                        }
                    }
                },
                _ => {
                    return fail_incorrectusage_clierror!(
                        "Invalid --cache-schema value: {cache_schema}. Valid values are 0, 1, -1 \
                         and -2"
                    )
                },
            }

            Ok((reader.finish()?, create_schema))
        }

        // ============ START OF NEW_JOIN MAIN CODE ==============
        let debuglog_flag = log::log_enabled!(log::Level::Debug);

        let delim = if let Some(delimiter) = self.flag_delimiter {
            delimiter.as_byte()
        } else {
            b','
        };

        let comment_char = if let Ok(comment_char) = env::var("QSV_COMMENT_CHAR") {
            Some(PlSmallStr::from_string(comment_char))
        } else {
            None
        };

        // Check if input files exist
        let mut input1_path = PathBuf::from(&self.arg_input1);
        if !input1_path.exists() {
            return fail_clierror!("Input file {} does not exist.", self.arg_input1);
        }
        let mut input2_path = PathBuf::from(&self.arg_input2);
        if !input2_path.exists() {
            return fail_clierror!("Input file {} does not exist.", self.arg_input2);
        }

        // Handle snappy compression for left input
        if input1_path.extension().and_then(std::ffi::OsStr::to_str) == Some("sz") {
            let decompressed_path = util::decompress_snappy_file(&input1_path, tmpdir)?;
            self.arg_input1.clone_from(&decompressed_path);
            input1_path = PathBuf::from(decompressed_path);
        }

        // Setup left LazyFrame
        let (mut left_lf, create_left_schema) = setup_lazy_frame(
            &input1_path,
            comment_char.as_ref(),
            self,
            delim,
            debuglog_flag,
        )?;

        if create_left_schema {
            let schema = left_lf.collect_schema()?;
            let schema_json = serde_json::to_string_pretty(&schema)?;
            let schema_file = input1_path.canonicalize()?.with_extension("pschema.json");
            let mut file = BufWriter::new(File::create(&schema_file)?);
            file.write_all(schema_json.as_bytes())?;
            file.flush()?;
            if debuglog_flag {
                log::debug!("Saved left schema to file: {}", schema_file.display());
            }
        }

        // Apply left filter if needed
        if let Some(filter_left) = &self.flag_filter_left {
            let filter_left_expr = polars::sql::sql_expr(filter_left)?;
            left_lf = left_lf.filter(filter_left_expr);
        }

        // Handle snappy compression for right input
        if input2_path.extension().and_then(std::ffi::OsStr::to_str) == Some("sz") {
            let decompressed_path = util::decompress_snappy_file(&input2_path, tmpdir)?;
            self.arg_input2.clone_from(&decompressed_path);
            input2_path = PathBuf::from(decompressed_path);
        }

        // Setup right LazyFrame
        let (mut right_lf, create_right_schema) = setup_lazy_frame(
            &input2_path,
            comment_char.as_ref(),
            self,
            delim,
            debuglog_flag,
        )?;

        if create_right_schema {
            let schema = right_lf.collect_schema()?;
            let schema_json = serde_json::to_string_pretty(&schema)?;
            let schema_file = input2_path.canonicalize()?.with_extension("pschema.json");
            let mut file = BufWriter::new(File::create(&schema_file)?);
            file.write_all(schema_json.as_bytes())?;
            file.flush()?;
            if debuglog_flag {
                log::debug!("Saved right schema to file: {}", schema_file.display());
            }
        }

        // Apply right filter if needed
        if let Some(filter_right) = &self.flag_filter_right {
            let filter_right_expr = polars::sql::sql_expr(filter_right)?;
            right_lf = right_lf.filter(filter_right_expr);
        }

        if self.flag_ignore_leading_zeros {
            // Transform the join keys in the left and right dataframe to handle leading zeros
            // For each join key column:
            //   1. Cast to string type
            //   2. If ignore_case is true, convert to lowercase
            //   3. Trim leading zeros, regardless of the original type
            // This allows joins to match on values like "001" and "1", "0001ABZ" and "1ABZ".
            // Note that the output columns will not have leading zeros for the join keys.
            //
            // We had to add ignore_case handling to ignore_leading_zeros in the join keys because
            // the existing code for ignore_case was not working for leading zeros if they are both
            // used at the same time.
            // The ignore_case code was creating temporary join key columns while
            // ignore_leading_zeros was using the existing join key columns.
            let ignore_case = self.flag_ignore_case;
            left_lf = left_lf.with_columns(
                self.arg_columns1
                    .split(',')
                    .map(|col| {
                        let col_name = col.to_string();
                        pl::col(&col_name).cast(pl::DataType::String).map(
                            move |s| {
                                Ok(Some(
                                    pl::Series::new(
                                        col_name.clone().into(),
                                        s.str()?
                                            .into_iter()
                                            .map(|x| {
                                                x.map(|val| {
                                                    let v = if ignore_case {
                                                        val.to_lowercase()
                                                    } else {
                                                        val.to_string()
                                                    };
                                                    if v.starts_with('0') {
                                                        v.trim_start_matches('0').to_string()
                                                    } else {
                                                        v
                                                    }
                                                })
                                                .unwrap_or_default()
                                            })
                                            .collect::<Vec<String>>(),
                                    )
                                    .into(),
                                ))
                            },
                            pl::GetOutput::from_type(pl::DataType::String),
                        )
                    })
                    .collect::<Vec<_>>(),
            );

            right_lf = right_lf.with_columns(
                self.arg_columns2
                    .split(',')
                    .map(|col| {
                        let col_name = col.to_string();
                        pl::col(&col_name).cast(pl::DataType::String).map(
                            move |s| {
                                Ok(Some(
                                    pl::Series::new(
                                        col_name.clone().into(),
                                        s.str()?
                                            .into_iter()
                                            .map(|x| {
                                                x.map(|val| {
                                                    let v = if ignore_case {
                                                        val.to_lowercase()
                                                    } else {
                                                        val.to_string()
                                                    };
                                                    if v.starts_with('0') {
                                                        v.trim_start_matches('0').to_string()
                                                    } else {
                                                        v
                                                    }
                                                })
                                                .unwrap_or_default()
                                            })
                                            .collect::<Vec<String>>(),
                                    )
                                    .into(),
                                ))
                            },
                            pl::GetOutput::from_type(pl::DataType::String),
                        )
                    })
                    .collect::<Vec<_>>(),
            );
        }

        Ok(JoinStruct {
            left_lf,
            left_sel: self.arg_columns1.clone(),
            right_lf,
            right_sel: self.arg_columns2.clone(),
            output: self.flag_output.clone(),
            delim,
            coalesce: self.flag_coalesce,
            streaming: self.flag_streaming,
            no_optimizations: self.flag_no_optimizations,
            sql_filter: self.flag_sql_filter.clone(),
            datetime_format: self.flag_datetime_format.clone(),
            date_format: self.flag_date_format.clone(),
            time_format: self.flag_time_format.clone(),
            float_precision: self.flag_float_precision,
            null_value: if self.flag_null_value == "<empty string>" {
                String::new()
            } else {
                self.flag_null_value.clone()
            },
            ignore_case: self.flag_ignore_case,
            ignore_leading_zeros: self.flag_ignore_leading_zeros,
        })
    }
}

/// if the file has a TSV/TAB or SSV extension, we automatically use
/// tab or semicolon as the delimiter
/// otherwise, we use the delimiter specified by the user
/// if the file has a .sz extension, we check the original file extension
/// to determine the delimiter
pub fn tsvssv_delim<P: AsRef<Path>>(file: P, orig_delim: u8) -> u8 {
    let inputfile_extension = file
        .as_ref()
        .extension()
        .and_then(std::ffi::OsStr::to_str)
        .unwrap_or_default()
        .to_ascii_lowercase();

    match inputfile_extension.as_str() {
        "tsv" | "tab" => b'\t',
        "ssv" => b';',
        "sz" => {
            // its a snappy compressed file
            // check what the original file extension is
            let orig_filestem = file
                .as_ref()
                .file_stem()
                .and_then(std::ffi::OsStr::to_str)
                .unwrap_or_default()
                .to_ascii_lowercase();
            let orig_extension = Path::new(&orig_filestem).extension().unwrap_or_default();
            if orig_extension == "tsv" || orig_extension == "tab" {
                b'\t'
            } else if orig_extension == "ssv" {
                b';'
            } else {
                orig_delim
            }
        },
        _ => orig_delim,
    }
}

#[test]
fn test_tsvssv_delim() {
    assert_eq!(tsvssv_delim("test.tsv", b','), b'\t');
    assert_eq!(tsvssv_delim("test.tab", b','), b'\t');
    assert_eq!(tsvssv_delim("test.ssv", b','), b';');
    assert_eq!(tsvssv_delim("test.sz", b','), b',');
    assert_eq!(tsvssv_delim("test.csv", b','), b',');
    assert_eq!(tsvssv_delim("test.TSV", b','), b'\t');
    assert_eq!(tsvssv_delim("test.Tab", b','), b'\t');
    assert_eq!(tsvssv_delim("test.SSV", b','), b';');
    assert_eq!(tsvssv_delim("test.sZ", b','), b',');
    assert_eq!(tsvssv_delim("test.CsV", b','), b',');
    assert_eq!(tsvssv_delim("test", b','), b',');
    assert_eq!(tsvssv_delim("test.csv.sz", b','), b',');
    assert_eq!(tsvssv_delim("test.tsv.sz", b','), b'\t');
    assert_eq!(tsvssv_delim("test.tab.sz", b','), b'\t');
    assert_eq!(tsvssv_delim("test.ssv.sz", b','), b';');
    assert_eq!(tsvssv_delim("test.csV.Sz", b','), b',');
    assert_eq!(tsvssv_delim("test.TSV.SZ", b','), b'\t');
    assert_eq!(tsvssv_delim("test.Tab.sZ", b','), b'\t');
    assert_eq!(tsvssv_delim("test.SSV.sz", b','), b';');
}