⚙️ API reference

class CoordinateSystemMismatchError(Exception):
    """Raised when two DataFrames have different coordinate systems.

    This error occurs when attempting range operations (overlap, nearest, etc.)
    on DataFrames where one uses 0-based coordinates and the other uses 1-based
    coordinates.

    Example:
        ```python
        df1 = pb.scan_vcf("file1.vcf", one_based=False)  # 0-based
        df2 = pb.scan_vcf("file2.vcf", one_based=True)   # 1-based
        pb.overlap(df1, df2)  # Raises CoordinateSystemMismatchError
        ```
    """

    pass

class MissingCoordinateSystemError(Exception):
    """Raised when a DataFrame lacks coordinate system metadata.

    Range operations require coordinate system metadata to determine the
    correct interval semantics. This error is raised when:

    - A Polars LazyFrame/DataFrame lacks polars-config-meta metadata
    - A Pandas DataFrame lacks df.attrs["coordinate_system_zero_based"]
    - A file path registers a table without Arrow schema metadata

    For Polars DataFrames, use polars-bio I/O functions (scan_*, read_*) which
    automatically set the metadata.

    For Pandas DataFrames, set the attribute before passing to range operations:
        ```python
        df.attrs["coordinate_system_zero_based"] = True  # 0-based coords
        ```

    Example:
        ```python
        import pandas as pd
        import polars_bio as pb

        pdf = pd.read_csv("intervals.bed", sep="\t", names=["chrom", "start", "end"])
        pb.overlap(pdf, pdf)  # Raises MissingCoordinateSystemError

        # Fix: set the coordinate system metadata
        pdf.attrs["coordinate_system_zero_based"] = True
        pb.overlap(pdf, pdf)  # Works correctly
        ```
    """

    pass

class IOOperations:
    @staticmethod
    def read_fasta(
        path: str,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
    ) -> pl.DataFrame:
        """

        Read a FASTA file into a DataFrame.

        Parameters:
            path: The path to the FASTA file.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the FASTA file. If not specified, it will be detected automatically based on the file extension. BGZF and GZIP compressions are supported ('bgz', 'gz').
            projection_pushdown: Enable column projection pushdown optimization. When True, only requested columns are processed at the DataFusion execution level, improving performance and reducing memory usage.

        !!! Example
            ```shell
            wget https://www.ebi.ac.uk/ena/browser/api/fasta/BK006935.2?download=true -O /tmp/test.fasta
            ```

            ```python
            import polars_bio as pb
            pb.read_fasta("/tmp/test.fasta").limit(1)
            ```
            ```shell
             shape: (1, 3)
            ┌─────────────────────────┬─────────────────────────────────┬─────────────────────────────────┐
            │ name                    ┆ description                     ┆ sequence                        │
            │ ---                     ┆ ---                             ┆ ---                             │
            │ str                     ┆ str                             ┆ str                             │
            ╞═════════════════════════╪═════════════════════════════════╪═════════════════════════════════╡
            │ ENA|BK006935|BK006935.2 ┆ TPA_inf: Saccharomyces cerevis… ┆ CCACACCACACCCACACACCCACACACCAC… │
            └─────────────────────────┴─────────────────────────────────┴─────────────────────────────────┘
            ```
        """
        return IOOperations.scan_fasta(
            path,
            chunk_size,
            concurrent_fetches,
            allow_anonymous,
            enable_request_payer,
            max_retries,
            timeout,
            compression_type,
            projection_pushdown,
        ).collect()

    @staticmethod
    def scan_fasta(
        path: str,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
    ) -> pl.LazyFrame:
        """

        Lazily read a FASTA file into a LazyFrame.

        Parameters:
            path: The path to the FASTA file.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the FASTA file. If not specified, it will be detected automatically based on the file extension. BGZF and GZIP compressions are supported ('bgz', 'gz').
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.

        !!! Example
            ```shell
            wget https://www.ebi.ac.uk/ena/browser/api/fasta/BK006935.2?download=true -O /tmp/test.fasta
            ```

            ```python
            import polars_bio as pb
            pb.scan_fasta("/tmp/test.fasta").limit(1).collect()
            ```
            ```shell
             shape: (1, 3)
            ┌─────────────────────────┬─────────────────────────────────┬─────────────────────────────────┐
            │ name                    ┆ description                     ┆ sequence                        │
            │ ---                     ┆ ---                             ┆ ---                             │
            │ str                     ┆ str                             ┆ str                             │
            ╞═════════════════════════╪═════════════════════════════════╪═════════════════════════════════╡
            │ ENA|BK006935|BK006935.2 ┆ TPA_inf: Saccharomyces cerevis… ┆ CCACACCACACCCACACACCCACACACCAC… │
            └─────────────────────────┴─────────────────────────────────┴─────────────────────────────────┘
            ```
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            max_retries=max_retries,
            timeout=timeout,
            compression_type=compression_type,
        )
        fasta_read_options = FastaReadOptions(
            object_storage_options=object_storage_options
        )
        read_options = ReadOptions(fasta_read_options=fasta_read_options)
        return _read_file(path, InputFormat.Fasta, read_options, projection_pushdown)

    @staticmethod
    def read_vcf(
        path: str,
        info_fields: Union[list[str], None] = None,
        format_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Read a VCF file into a DataFrame.

        !!! hint "Parallelism & Indexed Reads"
            Indexed parallel reads and predicate pushdown are automatic when a TBI/CSI index
            is present. See [File formats support](/polars-bio/features/#file-formats-support),
            [Indexed reads](/polars-bio/features/#indexed-reads-predicate-pushdown),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details.

        Parameters:
            path: The path to the VCF file.
            info_fields: List of INFO field names to include. If *None*, all INFO fields will be detected automatically from the VCF header. Use this to limit fields for better performance.
            format_fields: List of FORMAT field names to include (per-sample genotype data). If *None*, all FORMAT fields will be automatically detected from the VCF header. Column naming depends on the number of samples: for **single-sample** VCFs, columns are named directly by the FORMAT field (e.g., `GT`, `DP`); for **multi-sample** VCFs, columns are named `{sample_name}_{format_field}` (e.g., `NA12878_GT`, `NA12879_DP`). The GT field is always converted to string with `/` (unphased) or `|` (phased) separator.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the VCF file. If not specified, it will be detected automatically..
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            predicate_pushdown: Enable predicate pushdown using index files (TBI/CSI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.vcf.gz.tbi`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.

        !!! Example "Reading VCF with INFO and FORMAT fields"
            ```python
            import polars_bio as pb

            # Read VCF with both INFO and FORMAT fields
            df = pb.read_vcf(
                "sample.vcf.gz",
                info_fields=["END"],              # INFO field
                format_fields=["GT", "DP", "GQ"]  # FORMAT fields
            )

            # Single-sample VCF: FORMAT columns named directly (GT, DP, GQ)
            print(df.select(["chrom", "start", "ref", "alt", "END", "GT", "DP", "GQ"]))
            # Output:
            # shape: (10, 8)
            # ┌───────┬───────┬─────┬─────┬──────┬─────┬─────┬─────┐
            # │ chrom ┆ start ┆ ref ┆ alt ┆ END  ┆ GT  ┆ DP  ┆ GQ  │
            # │ str   ┆ u32   ┆ str ┆ str ┆ i32  ┆ str ┆ i32 ┆ i32 │
            # ╞═══════╪═══════╪═════╪═════╪══════╪═════╪═════╪═════╡
            # │ 1     ┆ 10009 ┆ A   ┆ .   ┆ null ┆ 0/0 ┆ 10  ┆ 27  │
            # │ 1     ┆ 10015 ┆ A   ┆ .   ┆ null ┆ 0/0 ┆ 17  ┆ 35  │
            # └───────┴───────┴─────┴─────┴──────┴─────┴─────┴─────┘

            # Multi-sample VCF: FORMAT columns named {sample}_{field}
            df = pb.read_vcf("multisample.vcf", format_fields=["GT", "DP"])
            print(df.select(["chrom", "start", "NA12878_GT", "NA12878_DP", "NA12879_GT"]))
            ```
        """
        lf = IOOperations.scan_vcf(
            path,
            info_fields,
            format_fields,
            chunk_size,
            concurrent_fetches,
            allow_anonymous,
            enable_request_payer,
            max_retries,
            timeout,
            compression_type,
            projection_pushdown,
            predicate_pushdown,
            use_zero_based,
        )
        # Get metadata before collecting (polars-config-meta doesn't preserve through collect)
        zero_based = lf.config_meta.get_metadata().get("coordinate_system_zero_based")
        df = lf.collect()
        # Set metadata on the collected DataFrame
        if zero_based is not None:
            set_coordinate_system(df, zero_based)
        return df

    @staticmethod
    def scan_vcf(
        path: str,
        info_fields: Union[list[str], None] = None,
        format_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.LazyFrame:
        """
        Lazily read a VCF file into a LazyFrame.

        !!! hint "Parallelism & Indexed Reads"
            Indexed parallel reads and predicate pushdown are automatic when a TBI/CSI index
            is present. See [File formats support](/polars-bio/features/#file-formats-support),
            [Indexed reads](/polars-bio/features/#indexed-reads-predicate-pushdown),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details.

        Parameters:
            path: The path to the VCF file.
            info_fields: List of INFO field names to include. If *None*, all INFO fields will be detected automatically from the VCF header. Use this to limit fields for better performance.
            format_fields: List of FORMAT field names to include (per-sample genotype data). If *None*, all FORMAT fields will be automatically detected from the VCF header. Column naming depends on the number of samples: for **single-sample** VCFs, columns are named directly by the FORMAT field (e.g., `GT`, `DP`); for **multi-sample** VCFs, columns are named `{sample_name}_{format_field}` (e.g., `NA12878_GT`, `NA12879_DP`). The GT field is always converted to string with `/` (unphased) or `|` (phased) separator.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the VCF file. If not specified, it will be detected automatically..
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            predicate_pushdown: Enable predicate pushdown using index files (TBI/CSI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.vcf.gz.tbi`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.

        !!! Example "Lazy scanning VCF with INFO and FORMAT fields"
            ```python
            import polars_bio as pb

            # Lazily scan VCF with both INFO and FORMAT fields
            lf = pb.scan_vcf(
                "sample.vcf.gz",
                info_fields=["END"],              # INFO field
                format_fields=["GT", "DP", "GQ"]  # FORMAT fields
            )

            # Apply filters and collect only what's needed
            df = lf.filter(pl.col("DP") > 20).select(
                ["chrom", "start", "ref", "alt", "GT", "DP", "GQ"]
            ).collect()

            # Single-sample VCF: FORMAT columns named directly (GT, DP, GQ)
            # Multi-sample VCF: FORMAT columns named {sample}_{field}
            ```
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            max_retries=max_retries,
            timeout=timeout,
            compression_type=compression_type,
        )

        # Use provided info_fields or autodetect from VCF header
        if info_fields is not None:
            initial_info_fields = info_fields
        else:
            # Get all info fields from VCF header for proper projection pushdown
            all_info_fields = None
            try:
                vcf_schema_df = IOOperations.describe_vcf(
                    path,
                    allow_anonymous=allow_anonymous,
                    enable_request_payer=enable_request_payer,
                    compression_type=compression_type,
                )
                # Use column name 'name' not 'id' based on the schema output
                all_info_fields = vcf_schema_df.select("name").to_series().to_list()
            except Exception:
                # Fallback to None if unable to get info fields
                all_info_fields = None

            # Always start with all info fields to establish full schema
            # The callback will re-register with only requested info fields for optimization
            initial_info_fields = all_info_fields

        zero_based = _resolve_zero_based(use_zero_based)
        vcf_read_options = VcfReadOptions(
            info_fields=initial_info_fields,
            format_fields=format_fields,
            object_storage_options=object_storage_options,
            zero_based=zero_based,
        )
        read_options = ReadOptions(vcf_read_options=vcf_read_options)
        return _read_file(
            path,
            InputFormat.Vcf,
            read_options,
            projection_pushdown,
            predicate_pushdown,
            zero_based=zero_based,
        )

    @staticmethod
    def read_gff(
        path: str,
        attr_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Read a GFF file into a DataFrame.

        Parameters:
            path: The path to the GFF file.
            attr_fields: List of attribute field names to extract as separate columns. If *None*, attributes will be kept as a nested structure. Use this to extract specific attributes like 'ID', 'gene_name', 'gene_type', etc. as direct columns for easier access.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the GFF file. If not specified, it will be detected automatically..
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            predicate_pushdown: Enable predicate pushdown using index files (TBI/CSI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.gff.gz.tbi`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.
        """
        lf = IOOperations.scan_gff(
            path,
            attr_fields,
            chunk_size,
            concurrent_fetches,
            allow_anonymous,
            enable_request_payer,
            max_retries,
            timeout,
            compression_type,
            projection_pushdown,
            predicate_pushdown,
            use_zero_based,
        )
        # Get metadata before collecting (polars-config-meta doesn't preserve through collect)
        zero_based = lf.config_meta.get_metadata().get("coordinate_system_zero_based")
        df = lf.collect()
        # Set metadata on the collected DataFrame
        if zero_based is not None:
            set_coordinate_system(df, zero_based)
        return df

    @staticmethod
    def scan_gff(
        path: str,
        attr_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.LazyFrame:
        """
        Lazily read a GFF file into a LazyFrame.

        Parameters:
            path: The path to the GFF file.
            attr_fields: List of attribute field names to extract as separate columns. If *None*, attributes will be kept as a nested structure. Use this to extract specific attributes like 'ID', 'gene_name', 'gene_type', etc. as direct columns for easier access.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large-scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the GFF file. If not specified, it will be detected automatically.
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            predicate_pushdown: Enable predicate pushdown using index files (TBI/CSI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.gff.gz.tbi`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            max_retries=max_retries,
            timeout=timeout,
            compression_type=compression_type,
        )

        zero_based = _resolve_zero_based(use_zero_based)
        gff_read_options = GffReadOptions(
            attr_fields=attr_fields,
            object_storage_options=object_storage_options,
            zero_based=zero_based,
        )
        read_options = ReadOptions(gff_read_options=gff_read_options)
        return _read_file(
            path,
            InputFormat.Gff,
            read_options,
            projection_pushdown,
            predicate_pushdown,
            zero_based=zero_based,
        )

    @staticmethod
    def read_bam(
        path: str,
        tag_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Read a BAM file into a DataFrame.

        !!! hint "Parallelism & Indexed Reads"
            Indexed parallel reads and predicate pushdown are automatic when a BAI/CSI index
            is present. See [File formats support](/polars-bio/features/#file-formats-support),
            [Indexed reads](/polars-bio/features/#indexed-reads-predicate-pushdown),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details.

        Parameters:
            path: The path to the BAM file.
            tag_fields: List of BAM tag names to include as columns (e.g., ["NM", "MD", "AS"]). If None, no optional tags are parsed (default). Common tags include: NM (edit distance), MD (mismatch string), AS (alignment score), XS (secondary alignment score), RG (read group), CB (cell barcode), UB (UMI barcode).
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large-scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large-scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            predicate_pushdown: Enable predicate pushdown using index files (BAI/CSI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.bam.bai`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.
        """
        lf = IOOperations.scan_bam(
            path,
            tag_fields,
            chunk_size,
            concurrent_fetches,
            allow_anonymous,
            enable_request_payer,
            max_retries,
            timeout,
            projection_pushdown,
            predicate_pushdown,
            use_zero_based,
        )
        # Get metadata before collecting (polars-config-meta doesn't preserve through collect)
        zero_based = lf.config_meta.get_metadata().get("coordinate_system_zero_based")
        df = lf.collect()
        # Set metadata on the collected DataFrame
        if zero_based is not None:
            set_coordinate_system(df, zero_based)
        return df

    @staticmethod
    def scan_bam(
        path: str,
        tag_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.LazyFrame:
        """
        Lazily read a BAM file into a LazyFrame.

        !!! hint "Parallelism & Indexed Reads"
            Indexed parallel reads and predicate pushdown are automatic when a BAI/CSI index
            is present. See [File formats support](/polars-bio/features/#file-formats-support),
            [Indexed reads](/polars-bio/features/#indexed-reads-predicate-pushdown),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details.

        Parameters:
            path: The path to the BAM file.
            tag_fields: List of BAM tag names to include as columns (e.g., ["NM", "MD", "AS"]). If None, no optional tags are parsed (default). Common tags include: NM (edit distance), MD (mismatch string), AS (alignment score), XS (secondary alignment score), RG (read group), CB (cell barcode), UB (UMI barcode).
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            predicate_pushdown: Enable predicate pushdown using index files (BAI/CSI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.bam.bai`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            max_retries=max_retries,
            timeout=timeout,
            compression_type="auto",
        )

        zero_based = _resolve_zero_based(use_zero_based)
        bam_read_options = BamReadOptions(
            object_storage_options=object_storage_options,
            zero_based=zero_based,
            tag_fields=tag_fields,
        )
        read_options = ReadOptions(bam_read_options=bam_read_options)
        return _read_file(
            path,
            InputFormat.Bam,
            read_options,
            projection_pushdown,
            predicate_pushdown,
            zero_based=zero_based,
        )

    @staticmethod
    def read_cram(
        path: str,
        reference_path: str = None,
        tag_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Read a CRAM file into a DataFrame.

        !!! hint "Parallelism & Indexed Reads"
            Indexed parallel reads and predicate pushdown are automatic when a CRAI index
            is present. See [File formats support](/polars-bio/features/#file-formats-support),
            [Indexed reads](/polars-bio/features/#indexed-reads-predicate-pushdown),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details.

        Parameters:
            path: The path to the CRAM file (local or cloud storage: S3, GCS, Azure Blob).
            reference_path: Optional path to external FASTA reference file (**local path only**, cloud storage not supported). If not provided, the CRAM file must contain embedded reference sequences. The FASTA file must have an accompanying index file (.fai) in the same directory. Create the index using: `samtools faidx reference.fasta`
            tag_fields: List of CRAM tag names to include as columns (e.g., ["NM", "MD", "AS"]). If None, no optional tags are parsed (default). Common tags include: NM (edit distance), MD (mismatch string), AS (alignment score), XS (secondary alignment score), RG (read group), CB (cell barcode), UB (UMI barcode).
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries: The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            projection_pushdown: Enable column projection pushdown optimization. When True, only requested columns are processed at the DataFusion execution level, improving performance and reducing memory usage.
            predicate_pushdown: Enable predicate pushdown using index files (CRAI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.cram.crai`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.

        !!! warning "Known Limitation: MD and NM Tags"
            Due to a limitation in the underlying noodles-cram library, **MD (mismatch descriptor) and NM (edit distance) tags are not accessible** from CRAM files, even when stored in the file. These tags can be seen with samtools but are not exposed through the noodles-cram record.data() interface.

            Other optional tags (RG, MQ, AM, OQ, etc.) work correctly. This issue is tracked at: https://github.com/biodatageeks/datafusion-bio-formats/issues/54

            **Workaround**: Use BAM format if MD/NM tags are required for your analysis.

        !!! example "Using External Reference"
            ```python
            import polars_bio as pb

            # Read CRAM with external reference
            df = pb.read_cram(
                "/path/to/file.cram",
                reference_path="/path/to/reference.fasta"
            )
            ```

        !!! example "Public CRAM File Example"
            Download and read a public CRAM file from 42basepairs:
            ```bash
            # Download the CRAM file and reference
            wget https://42basepairs.com/download/s3/gatk-test-data/wgs_cram/NA12878_20k_hg38/NA12878.cram
            wget https://storage.googleapis.com/genomics-public-data/resources/broad/hg38/v0/Homo_sapiens_assembly38.fasta

            # Create FASTA index (required)
            samtools faidx Homo_sapiens_assembly38.fasta
            ```

            ```python
            import polars_bio as pb

            # Read first 5 reads from the CRAM file
            df = pb.scan_cram(
                "NA12878.cram",
                reference_path="Homo_sapiens_assembly38.fasta"
            ).limit(5).collect()

            print(df.select(["name", "chrom", "start", "end", "cigar"]))
            ```

        !!! example "Creating CRAM with Embedded Reference"
            To create a CRAM file with embedded reference using samtools:
            ```bash
            samtools view -C -o output.cram --output-fmt-option embed_ref=1 input.bam
            ```

        Returns:
            A Polars DataFrame with the following schema:
                - name: Read name (String)
                - chrom: Chromosome/contig name (String)
                - start: Alignment start position, 1-based (UInt32)
                - end: Alignment end position, 1-based (UInt32)
                - flags: SAM flags (UInt32)
                - cigar: CIGAR string (String)
                - mapping_quality: Mapping quality (UInt32)
                - mate_chrom: Mate chromosome/contig name (String)
                - mate_start: Mate alignment start position, 1-based (UInt32)
                - sequence: Read sequence (String)
                - quality_scores: Base quality scores (String)
        """
        lf = IOOperations.scan_cram(
            path,
            reference_path,
            tag_fields,
            chunk_size,
            concurrent_fetches,
            allow_anonymous,
            enable_request_payer,
            max_retries,
            timeout,
            projection_pushdown,
            predicate_pushdown,
            use_zero_based,
        )
        # Get metadata before collecting (polars-config-meta doesn't preserve through collect)
        zero_based = lf.config_meta.get_metadata().get("coordinate_system_zero_based")
        df = lf.collect()
        # Set metadata on the collected DataFrame
        if zero_based is not None:
            set_coordinate_system(df, zero_based)
        return df

    @staticmethod
    def scan_cram(
        path: str,
        reference_path: str = None,
        tag_fields: Union[list[str], None] = None,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        projection_pushdown: bool = True,
        predicate_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.LazyFrame:
        """
        Lazily read a CRAM file into a LazyFrame.

        !!! hint "Parallelism & Indexed Reads"
            Indexed parallel reads and predicate pushdown are automatic when a CRAI index
            is present. See [File formats support](/polars-bio/features/#file-formats-support),
            [Indexed reads](/polars-bio/features/#indexed-reads-predicate-pushdown),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details.

        Parameters:
            path: The path to the CRAM file (local or cloud storage: S3, GCS, Azure Blob).
            reference_path: Optional path to external FASTA reference file (**local path only**, cloud storage not supported). If not provided, the CRAM file must contain embedded reference sequences. The FASTA file must have an accompanying index file (.fai) in the same directory. Create the index using: `samtools faidx reference.fasta`
            tag_fields: List of CRAM tag names to include as columns (e.g., ["NM", "MD", "AS"]). If None, no optional tags are parsed (default). Common tags include: NM (edit distance), MD (mismatch string), AS (alignment score), XS (secondary alignment score), RG (read group), CB (cell barcode), UB (UMI barcode).
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries: The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            projection_pushdown: Enable column projection pushdown optimization. When True, only requested columns are processed at the DataFusion execution level, improving performance and reducing memory usage.
            predicate_pushdown: Enable predicate pushdown using index files (CRAI) for efficient region-based filtering. Index files are auto-discovered (e.g., `file.cram.crai`). Only simple predicates are pushed down (equality, comparisons, IN); complex predicates like `.str.contains()` or OR logic are filtered client-side. Correctness is always guaranteed.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.

        !!! warning "Known Limitation: MD and NM Tags"
            Due to a limitation in the underlying noodles-cram library, **MD (mismatch descriptor) and NM (edit distance) tags are not accessible** from CRAM files, even when stored in the file. These tags can be seen with samtools but are not exposed through the noodles-cram record.data() interface.

            Other optional tags (RG, MQ, AM, OQ, etc.) work correctly. This issue is tracked at: https://github.com/biodatageeks/datafusion-bio-formats/issues/54

            **Workaround**: Use BAM format if MD/NM tags are required for your analysis.

        !!! example "Using External Reference"
            ```python
            import polars_bio as pb

            # Lazy scan CRAM with external reference
            lf = pb.scan_cram(
                "/path/to/file.cram",
                reference_path="/path/to/reference.fasta"
            )

            # Apply transformations and collect
            df = lf.filter(pl.col("chrom") == "chr1").collect()
            ```

        !!! example "Public CRAM File Example"
            Download and read a public CRAM file from 42basepairs:
            ```bash
            # Download the CRAM file and reference
            wget https://42basepairs.com/download/s3/gatk-test-data/wgs_cram/NA12878_20k_hg38/NA12878.cram
            wget https://storage.googleapis.com/genomics-public-data/resources/broad/hg38/v0/Homo_sapiens_assembly38.fasta

            # Create FASTA index (required)
            samtools faidx Homo_sapiens_assembly38.fasta
            ```

            ```python
            import polars_bio as pb
            import polars as pl

            # Lazy scan and filter for chromosome 20 reads
            df = pb.scan_cram(
                "NA12878.cram",
                reference_path="Homo_sapiens_assembly38.fasta"
            ).filter(
                pl.col("chrom") == "chr20"
            ).select(
                ["name", "chrom", "start", "end", "mapping_quality"]
            ).limit(10).collect()

            print(df)
            ```

        !!! example "Creating CRAM with Embedded Reference"
            To create a CRAM file with embedded reference using samtools:
            ```bash
            samtools view -C -o output.cram --output-fmt-option embed_ref=1 input.bam
            ```

        Returns:
            A Polars LazyFrame with the following schema:
                - name: Read name (String)
                - chrom: Chromosome/contig name (String)
                - start: Alignment start position, 1-based (UInt32)
                - end: Alignment end position, 1-based (UInt32)
                - flags: SAM flags (UInt32)
                - cigar: CIGAR string (String)
                - mapping_quality: Mapping quality (UInt32)
                - mate_chrom: Mate chromosome/contig name (String)
                - mate_start: Mate alignment start position, 1-based (UInt32)
                - sequence: Read sequence (String)
                - quality_scores: Base quality scores (String)
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            max_retries=max_retries,
            timeout=timeout,
            compression_type="auto",
        )

        zero_based = _resolve_zero_based(use_zero_based)
        cram_read_options = CramReadOptions(
            reference_path=reference_path,
            object_storage_options=object_storage_options,
            zero_based=zero_based,
            tag_fields=tag_fields,
        )
        read_options = ReadOptions(cram_read_options=cram_read_options)
        return _read_file(
            path,
            InputFormat.Cram,
            read_options,
            projection_pushdown,
            predicate_pushdown,
            zero_based=zero_based,
        )

    @staticmethod
    def describe_bam(
        path: str,
        sample_size: int = 100,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Get schema information for a BAM file with automatic tag discovery.

        Samples the first N records to discover all available tags and their types.
        Returns detailed schema information including column names, data types,
        nullability, category (standard/tag), SAM type, and descriptions.

        Parameters:
            path: The path to the BAM file.
            sample_size: Number of records to sample for tag discovery (default: 100).
                Use higher values for more comprehensive tag discovery.
            chunk_size: The size in MB of a chunk when reading from object storage.
            concurrent_fetches: The number of concurrent fetches when reading from object storage.
            allow_anonymous: Whether to allow anonymous access to object storage.
            enable_request_payer: Whether to enable request payer for object storage.
            max_retries: The maximum number of retries for reading the file.
            timeout: The timeout in seconds for reading the file.
            compression_type: The compression type of the file. If "auto" (default), compression is detected automatically.
            use_zero_based: If True, output 0-based coordinates. If False, 1-based coordinates.

        Returns:
            DataFrame with columns:
            - column_name: Name of the column/field
            - data_type: Arrow data type (e.g., "Utf8", "Int32")
            - nullable: Whether the field can be null
            - category: "core" for fixed columns, "tag" for optional SAM tags
            - sam_type: SAM type code (e.g., "Z", "i") for tags, null for core columns
            - description: Human-readable description of the field

        Example:
            ```python
            import polars_bio as pb

            # Auto-discover all tags present in the file
            schema = pb.describe_bam("file.bam", sample_size=100)
            print(schema)
            # Output:
            # shape: (15, 6)
            # ┌─────────────┬───────────┬──────────┬──────────┬──────────┬──────────────────────┐
            # │ column_name ┆ data_type ┆ nullable ┆ category ┆ sam_type ┆ description          │
            # │ ---         ┆ ---       ┆ ---      ┆ ---      ┆ ---      ┆ ---                  │
            # │ str         ┆ str       ┆ bool     ┆ str      ┆ str      ┆ str                  │
            # ╞═════════════╪═══════════╪══════════╪══════════╪══════════╪══════════════════════╡
            # │ name        ┆ Utf8      ┆ true     ┆ core     ┆ null     ┆ Query name           │
            # │ chrom       ┆ Utf8      ┆ true     ┆ core     ┆ null     ┆ Reference name       │
            # │ ...         ┆ ...       ┆ ...      ┆ ...      ┆ ...      ┆ ...                  │
            # │ NM          ┆ Int32     ┆ true     ┆ tag      ┆ i        ┆ Edit distance        │
            # │ AS          ┆ Int32     ┆ true     ┆ tag      ┆ i        ┆ Alignment score      │
            # └─────────────┴───────────┴──────────┴──────────┴──────────┴──────────────────────┘
            ```
        """
        # Build object storage options
        object_storage_options = PyObjectStorageOptions(
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            max_retries=max_retries,
            timeout=timeout,
            compression_type=compression_type,
        )

        # Resolve zero_based setting
        zero_based = _resolve_zero_based(use_zero_based)

        # Call Rust function with tag auto-discovery (tag_fields=None)
        df = py_describe_bam(
            ctx,  # PyBioSessionContext
            path,
            object_storage_options,
            zero_based,
            None,  # tag_fields=None enables auto-discovery
            sample_size,
        )

        # Convert DataFusion DataFrame to Polars DataFrame
        return pl.from_arrow(df.to_arrow_table())

    @staticmethod
    def describe_cram(
        path: str,
        reference_path: str = None,
        sample_size: int = 100,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Get schema information for a CRAM file with automatic tag discovery.

        Samples the first N records to discover all available tags and their types.
        Returns detailed schema information including column names, data types,
        nullability, category (core/tag), SAM type, and descriptions.

        Parameters:
            path: The path to the CRAM file.
            reference_path: Optional path to external FASTA reference file.
            sample_size: Number of records to sample for tag discovery (default: 100).
            chunk_size: The size in MB of a chunk when reading from object storage.
            concurrent_fetches: The number of concurrent fetches when reading from object storage.
            allow_anonymous: Whether to allow anonymous access to object storage.
            enable_request_payer: Whether to enable request payer for object storage.
            max_retries: The maximum number of retries for reading the file.
            timeout: The timeout in seconds for reading the file.
            compression_type: The compression type of the file. If "auto" (default), compression is detected automatically.
            use_zero_based: If True, output 0-based coordinates. If False, 1-based coordinates.

        Returns:
            DataFrame with columns:
            - column_name: Name of the column/field
            - data_type: Arrow data type (e.g., "Utf8", "Int32")
            - nullable: Whether the field can be null
            - category: "core" for fixed columns, "tag" for optional SAM tags
            - sam_type: SAM type code (e.g., "Z", "i") for tags, null for core columns
            - description: Human-readable description of the field

        !!! warning "Known Limitation: MD and NM Tags"
            Due to a limitation in the underlying noodles-cram library, **MD (mismatch descriptor) and NM (edit distance) tags are not discoverable** from CRAM files, even when stored. Automatic tag discovery will not include MD/NM tags. Other optional tags (RG, MQ, AM, OQ, etc.) are discovered correctly. See: https://github.com/biodatageeks/datafusion-bio-formats/issues/54

        Example:
            ```python
            import polars_bio as pb

            # Auto-discover all tags present in the file
            schema = pb.describe_cram("file.cram", sample_size=100)
            print(schema)

            # Filter to see only tag columns
            tags = schema.filter(schema["category"] == "tag")
            print(tags["column_name"])
            ```
        """
        # Build object storage options
        object_storage_options = PyObjectStorageOptions(
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            max_retries=max_retries,
            timeout=timeout,
            compression_type=compression_type,
        )

        # Resolve zero_based setting
        zero_based = _resolve_zero_based(use_zero_based)

        # Call Rust function with tag auto-discovery (tag_fields=None)
        df = py_describe_cram(
            ctx,
            path,
            reference_path,
            object_storage_options,
            zero_based,
            None,  # tag_fields=None enables auto-discovery
            sample_size,
        )

        # Convert DataFusion DataFrame to Polars DataFrame
        return pl.from_arrow(df.to_arrow_table())

    @staticmethod
    def read_fastq(
        path: str,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
    ) -> pl.DataFrame:
        """
        Read a FASTQ file into a DataFrame.

        !!! hint "Parallelism & Compression"
            See [File formats support](/polars-bio/features/#file-formats-support),
            [Compression](/polars-bio/features/#compression),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details on parallel reads and supported compression types.

        Parameters:
            path: The path to the FASTQ file.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the FASTQ file. If not specified, it will be detected automatically based on the file extension. BGZF and GZIP compressions are supported ('bgz', 'gz').
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
        """
        return IOOperations.scan_fastq(
            path,
            chunk_size,
            concurrent_fetches,
            allow_anonymous,
            enable_request_payer,
            max_retries,
            timeout,
            compression_type,
            projection_pushdown,
        ).collect()

    @staticmethod
    def scan_fastq(
        path: str,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
    ) -> pl.LazyFrame:
        """
        Lazily read a FASTQ file into a LazyFrame.

        !!! hint "Parallelism & Compression"
            See [File formats support](/polars-bio/features/#file-formats-support),
            [Compression](/polars-bio/features/#compression),
            and [Automatic parallel partitioning](/polars-bio/features/#automatic-parallel-partitioning) for details on parallel reads and supported compression types.

        Parameters:
            path: The path to the FASTQ file.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the FASTQ file. If not specified, it will be detected automatically based on the file extension. BGZF and GZIP compressions are supported ('bgz', 'gz').
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            max_retries=max_retries,
            timeout=timeout,
            compression_type=compression_type,
        )

        fastq_read_options = FastqReadOptions(
            object_storage_options=object_storage_options,
        )
        read_options = ReadOptions(fastq_read_options=fastq_read_options)
        return _read_file(path, InputFormat.Fastq, read_options, projection_pushdown)

    @staticmethod
    def read_bed(
        path: str,
        thread_num: int = 1,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Read a BED file into a DataFrame.

        Parameters:
            path: The path to the BED file.
            thread_num: The number of threads to use for reading the BED file. Used **only** for parallel decompression of BGZF blocks. Works only for **local** files.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the BED file. If not specified, it will be detected automatically based on the file extension. BGZF compressions is supported ('bgz').
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! Note
            Only **BED4** format is supported. It extends the basic BED format (BED3) by adding a name field, resulting in four columns: chromosome, start position, end position, and name.
            Also unlike other text formats, **GZIP** compression is not supported.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.
        """
        lf = IOOperations.scan_bed(
            path,
            thread_num,
            chunk_size,
            concurrent_fetches,
            allow_anonymous,
            enable_request_payer,
            max_retries,
            timeout,
            compression_type,
            projection_pushdown,
            use_zero_based,
        )
        # Get metadata before collecting (polars-config-meta doesn't preserve through collect)
        zero_based = lf.config_meta.get_metadata().get("coordinate_system_zero_based")
        df = lf.collect()
        # Set metadata on the collected DataFrame
        if zero_based is not None:
            set_coordinate_system(df, zero_based)
        return df

    @staticmethod
    def scan_bed(
        path: str,
        thread_num: int = 1,
        chunk_size: int = 8,
        concurrent_fetches: int = 1,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        max_retries: int = 5,
        timeout: int = 300,
        compression_type: str = "auto",
        projection_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.LazyFrame:
        """
        Lazily read a BED file into a LazyFrame.

        Parameters:
            path: The path to the BED file.
            thread_num: The number of threads to use for reading the BED file. Used **only** for parallel decompression of BGZF blocks. Works only for **local** files.
            chunk_size: The size in MB of a chunk when reading from an object store. The default is 8 MB. For large scale operations, it is recommended to increase this value to 64.
            concurrent_fetches: [GCS] The number of concurrent fetches when reading from an object store. The default is 1. For large scale operations, it is recommended to increase this value to 8 or even more.
            allow_anonymous: [GCS, AWS S3] Whether to allow anonymous access to object storage.
            enable_request_payer: [AWS S3] Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            max_retries:  The maximum number of retries for reading the file from object storage.
            timeout: The timeout in seconds for reading the file from object storage.
            compression_type: The compression type of the BED file. If not specified, it will be detected automatically based on the file extension. BGZF compressions is supported ('bgz').
            projection_pushdown: Enable column projection pushdown to optimize query performance by only reading the necessary columns at the DataFusion level.
            use_zero_based: If True, output 0-based half-open coordinates. If False, output 1-based closed coordinates. If None (default), uses the global configuration `datafusion.bio.coordinate_system_zero_based`.

        !!! Note
            Only **BED4** format is supported. It extends the basic BED format (BED3) by adding a name field, resulting in four columns: chromosome, start position, end position, and name.
            Also unlike other text formats, **GZIP** compression is not supported.

        !!! note
            By default, coordinates are output in **1-based closed** format. Use `use_zero_based=True` or set `pb.set_option(pb.POLARS_BIO_COORDINATE_SYSTEM_ZERO_BASED, True)` for 0-based half-open coordinates.
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=chunk_size,
            concurrent_fetches=concurrent_fetches,
            max_retries=max_retries,
            timeout=timeout,
            compression_type=compression_type,
        )

        zero_based = _resolve_zero_based(use_zero_based)
        bed_read_options = BedReadOptions(
            thread_num=thread_num,
            object_storage_options=object_storage_options,
            zero_based=zero_based,
        )
        read_options = ReadOptions(bed_read_options=bed_read_options)
        return _read_file(
            path,
            InputFormat.Bed,
            read_options,
            projection_pushdown,
            zero_based=zero_based,
        )

    @staticmethod
    def read_table(path: str, schema: Dict = None, **kwargs) -> pl.DataFrame:
        """
         Read a tab-delimited (i.e. BED) file into a Polars DataFrame.
         Tries to be compatible with Bioframe's [read_table](https://bioframe.readthedocs.io/en/latest/guide-io.html)
         but faster. Schema should follow the Bioframe's schema [format](https://github.com/open2c/bioframe/blob/2b685eebef393c2c9e6220dcf550b3630d87518e/bioframe/io/schemas.py#L174).

        Parameters:
            path: The path to the file.
            schema: Schema should follow the Bioframe's schema [format](https://github.com/open2c/bioframe/blob/2b685eebef393c2c9e6220dcf550b3630d87518e/bioframe/io/schemas.py#L174).
        """
        return IOOperations.scan_table(path, schema, **kwargs).collect()

    @staticmethod
    def scan_table(path: str, schema: Dict = None, **kwargs) -> pl.LazyFrame:
        """
         Lazily read a tab-delimited (i.e. BED) file into a Polars LazyFrame.
         Tries to be compatible with Bioframe's [read_table](https://bioframe.readthedocs.io/en/latest/guide-io.html)
         but faster and lazy. Schema should follow the Bioframe's schema [format](https://github.com/open2c/bioframe/blob/2b685eebef393c2c9e6220dcf550b3630d87518e/bioframe/io/schemas.py#L174).

        Parameters:
            path: The path to the file.
            schema: Schema should follow the Bioframe's schema [format](https://github.com/open2c/bioframe/blob/2b685eebef393c2c9e6220dcf550b3630d87518e/bioframe/io/schemas.py#L174).
        """
        df = pl.scan_csv(path, separator="\t", has_header=False, **kwargs)
        if schema is not None:
            columns = SCHEMAS[schema]
            if len(columns) != len(df.collect_schema()):
                raise ValueError(
                    f"Schema incompatible with the input. Expected {len(columns)} columns in a schema, got {len(df.collect_schema())} in the input data file. Please provide a valid schema."
                )
            for i, c in enumerate(columns):
                df = df.rename({f"column_{i+1}": c})
        return df

    @staticmethod
    def describe_vcf(
        path: str,
        allow_anonymous: bool = True,
        enable_request_payer: bool = False,
        compression_type: str = "auto",
    ) -> pl.DataFrame:
        """
        Describe VCF INFO schema.

        Parameters:
            path: The path to the VCF file.
            allow_anonymous: Whether to allow anonymous access to object storage (GCS and S3 supported).
            enable_request_payer: Whether to enable request payer for object storage. This is useful for reading files from AWS S3 buckets that require request payer.
            compression_type: The compression type of the VCF file. If not specified, it will be detected automatically..
        """
        object_storage_options = PyObjectStorageOptions(
            allow_anonymous=allow_anonymous,
            enable_request_payer=enable_request_payer,
            chunk_size=8,
            concurrent_fetches=1,
            max_retries=1,
            timeout=10,
            compression_type=compression_type,
        )
        return py_describe_vcf(ctx, path, object_storage_options).to_polars()

    @staticmethod
    def from_polars(name: str, df: Union[pl.DataFrame, pl.LazyFrame]) -> None:
        """
        Register a Polars DataFrame as a DataFusion table.

        Parameters:
            name: The name of the table.
            df: The Polars DataFrame.
        """
        reader = (
            df.to_arrow()
            if isinstance(df, pl.DataFrame)
            else df.collect().to_arrow().to_reader()
        )
        py_from_polars(ctx, name, reader)

    @staticmethod
    def write_vcf(
        df: Union[pl.DataFrame, pl.LazyFrame],
        path: str,
    ) -> int:
        """
        Write a DataFrame to VCF format.

        Coordinate system is automatically read from DataFrame metadata (set during
        read_vcf). Compression is auto-detected from the file extension.

        Parameters:
            df: The DataFrame or LazyFrame to write.
            path: The output file path. Compression is auto-detected from extension
                  (.vcf.bgz for BGZF, .vcf.gz for GZIP, .vcf for uncompressed).

        Returns:
            The number of rows written.

        !!! Example "Writing VCF files"
            ```python
            import polars_bio as pb

            # Read a VCF file
            df = pb.read_vcf("input.vcf")

            # Write to uncompressed VCF
            pb.write_vcf(df, "output.vcf")

            # Write to BGZF-compressed VCF
            pb.write_vcf(df, "output.vcf.bgz")

            # Write to GZIP-compressed VCF
            pb.write_vcf(df, "output.vcf.gz")
            ```
        """
        return _write_file(df, path, OutputFormat.Vcf)

    @staticmethod
    def sink_vcf(
        lf: pl.LazyFrame,
        path: str,
    ) -> None:
        """
        Streaming write a LazyFrame to VCF format.

        This method executes the LazyFrame immediately and writes the results
        to the specified path. Unlike `write_vcf`, it doesn't return the row count.

        Coordinate system is automatically read from LazyFrame metadata (set during
        scan_vcf). Compression is auto-detected from the file extension.

        Parameters:
            lf: The LazyFrame to write.
            path: The output file path. Compression is auto-detected from extension
                  (.vcf.bgz for BGZF, .vcf.gz for GZIP, .vcf for uncompressed).

        !!! Example "Streaming write VCF"
            ```python
            import polars_bio as pb

            # Lazy read and filter, then sink to VCF
            lf = pb.scan_vcf("large_input.vcf").filter(pl.col("qual") > 30)
            pb.sink_vcf(lf, "filtered_output.vcf.bgz")
            ```
        """
        _write_file(lf, path, OutputFormat.Vcf)

    @staticmethod
    def write_fastq(
        df: Union[pl.DataFrame, pl.LazyFrame],
        path: str,
    ) -> int:
        """
        Write a DataFrame to FASTQ format.

        Compression is auto-detected from the file extension.

        Parameters:
            df: The DataFrame or LazyFrame to write. Must have columns:
                - name: Read name/identifier
                - sequence: DNA sequence
                - quality_scores: Quality scores string
                Optional: description (added after name on header line)
            path: The output file path. Compression is auto-detected from extension
                  (.fastq.bgz for BGZF, .fastq.gz for GZIP, .fastq for uncompressed).

        Returns:
            The number of rows written.

        !!! Example "Writing FASTQ files"
            ```python
            import polars_bio as pb

            # Read a FASTQ file
            df = pb.read_fastq("input.fastq")

            # Write to uncompressed FASTQ
            pb.write_fastq(df, "output.fastq")

            # Write to GZIP-compressed FASTQ
            pb.write_fastq(df, "output.fastq.gz")
            ```
        """
        return _write_file(df, path, OutputFormat.Fastq)

    @staticmethod
    def sink_fastq(
        lf: pl.LazyFrame,
        path: str,
    ) -> None:
        """
        Streaming write a LazyFrame to FASTQ format.

        Compression is auto-detected from the file extension.

        Parameters:
            lf: The LazyFrame to write.
            path: The output file path. Compression is auto-detected from extension
                  (.fastq.bgz for BGZF, .fastq.gz for GZIP, .fastq for uncompressed).

        !!! Example "Streaming write FASTQ"
            ```python
            import polars_bio as pb

            # Lazy read, filter by quality, then sink
            lf = pb.scan_fastq("large_input.fastq.gz")
            pb.sink_fastq(lf.limit(1000), "sample_output.fastq")
            ```
        """
        _write_file(lf, path, OutputFormat.Fastq)

    @staticmethod
    def write_bam(
        df: Union[pl.DataFrame, pl.LazyFrame],
        path: str,
        sort_on_write: bool = False,
    ) -> int:
        """
        Write a DataFrame to BAM/SAM format.

        Compression is auto-detected from file extension:
        - .sam → Uncompressed SAM (plain text)
        - .bam → BGZF-compressed BAM

        For CRAM format, use `write_cram()` instead.

        Parameters:
            df: DataFrame or LazyFrame with 11 core BAM columns + optional tag columns
            path: Output file path (.bam or .sam)
            sort_on_write: If True, sort records by (chrom, start) and set header SO:coordinate.
                If False (default), set header SO:unsorted.

        Returns:
            Number of rows written

        !!! Example "Write BAM files"
            ```python
            import polars_bio as pb
            df = pb.read_bam("input.bam", tag_fields=["NM", "AS"])
            pb.write_bam(df, "output.bam")
            pb.write_bam(df, "output.sam")
            ```
        """
        return _write_bam_file(
            df, path, OutputFormat.Bam, None, sort_on_write=sort_on_write
        )

    @staticmethod
    def sink_bam(
        lf: pl.LazyFrame,
        path: str,
        sort_on_write: bool = False,
    ) -> None:
        """
        Streaming write a LazyFrame to BAM/SAM format.

        For CRAM format, use `sink_cram()` instead.

        Parameters:
            lf: LazyFrame to write
            path: Output file path (.bam or .sam)
            sort_on_write: If True, sort records by (chrom, start) and set header SO:coordinate.
                If False (default), set header SO:unsorted.

        !!! Example "Streaming write BAM"
            ```python
            import polars_bio as pb
            lf = pb.scan_bam("input.bam").filter(pl.col("mapping_quality") > 20)
            pb.sink_bam(lf, "filtered.bam")
            ```
        """
        _write_bam_file(lf, path, OutputFormat.Bam, None, sort_on_write=sort_on_write)

    @staticmethod
    def read_sam(
        path: str,
        tag_fields: Union[list[str], None] = None,
        projection_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Read a SAM file into a DataFrame.

        SAM (Sequence Alignment/Map) is the plain-text counterpart of BAM.
        This function reuses the BAM reader, which auto-detects the format
        from the file extension.

        Parameters:
            path: The path to the SAM file.
            tag_fields: List of SAM tag names to include as columns (e.g., ["NM", "MD", "AS"]).
                If None, no optional tags are parsed (default).
            projection_pushdown: Enable column projection pushdown to optimize query performance.
            use_zero_based: If True, output 0-based half-open coordinates.
                If False, output 1-based closed coordinates.
                If None (default), uses the global configuration.

        !!! note
            By default, coordinates are output in **1-based closed** format.
        """
        lf = IOOperations.scan_sam(
            path,
            tag_fields,
            projection_pushdown,
            use_zero_based,
        )
        zero_based = lf.config_meta.get_metadata().get("coordinate_system_zero_based")
        df = lf.collect()
        if zero_based is not None:
            set_coordinate_system(df, zero_based)
        return df

    @staticmethod
    def scan_sam(
        path: str,
        tag_fields: Union[list[str], None] = None,
        projection_pushdown: bool = True,
        use_zero_based: Optional[bool] = None,
    ) -> pl.LazyFrame:
        """
        Lazily read a SAM file into a LazyFrame.

        SAM (Sequence Alignment/Map) is the plain-text counterpart of BAM.
        This function reuses the BAM reader, which auto-detects the format
        from the file extension.

        Parameters:
            path: The path to the SAM file.
            tag_fields: List of SAM tag names to include as columns (e.g., ["NM", "MD", "AS"]).
                If None, no optional tags are parsed (default).
            projection_pushdown: Enable column projection pushdown to optimize query performance.
            use_zero_based: If True, output 0-based half-open coordinates.
                If False, output 1-based closed coordinates.
                If None (default), uses the global configuration.

        !!! note
            By default, coordinates are output in **1-based closed** format.
        """
        zero_based = _resolve_zero_based(use_zero_based)
        bam_read_options = BamReadOptions(
            zero_based=zero_based,
            tag_fields=tag_fields,
        )
        read_options = ReadOptions(bam_read_options=bam_read_options)
        return _read_file(
            path,
            InputFormat.Sam,
            read_options,
            projection_pushdown,
            zero_based=zero_based,
        )

    @staticmethod
    def describe_sam(
        path: str,
        sample_size: int = 100,
        use_zero_based: Optional[bool] = None,
    ) -> pl.DataFrame:
        """
        Get schema information for a SAM file with automatic tag discovery.

        Samples the first N records to discover all available tags and their types.
        Reuses the BAM describe logic, which auto-detects SAM from the file extension.

        Parameters:
            path: The path to the SAM file.
            sample_size: Number of records to sample for tag discovery (default: 100).
            use_zero_based: If True, output 0-based coordinates. If False, 1-based coordinates.

        Returns:
            DataFrame with columns: column_name, data_type, nullable, category, sam_type, description
        """
        zero_based = _resolve_zero_based(use_zero_based)

        df = py_describe_bam(
            ctx,
            path,
            None,
            zero_based,
            None,
            sample_size,
        )

        return pl.from_arrow(df.to_arrow_table())

    @staticmethod
    def write_sam(
        df: Union[pl.DataFrame, pl.LazyFrame],
        path: str,
        sort_on_write: bool = False,
    ) -> int:
        """
        Write a DataFrame to SAM format (plain text).

        Parameters:
            df: DataFrame or LazyFrame with 11 core BAM/SAM columns + optional tag columns
            path: Output file path (.sam)
            sort_on_write: If True, sort records by (chrom, start) and set header SO:coordinate.
                If False (default), set header SO:unsorted.

        Returns:
            Number of rows written

        !!! Example "Write SAM files"
            ```python
            import polars_bio as pb
            df = pb.read_bam("input.bam", tag_fields=["NM", "AS"])
            pb.write_sam(df, "output.sam")
            ```
        """
        return _write_bam_file(
            df, path, OutputFormat.Sam, None, sort_on_write=sort_on_write
        )

    @staticmethod
    def sink_sam(
        lf: pl.LazyFrame,
        path: str,
        sort_on_write: bool = False,
    ) -> None:
        """
        Streaming write a LazyFrame to SAM format (plain text).

        Parameters:
            lf: LazyFrame to write
            path: Output file path (.sam)
            sort_on_write: If True, sort records by (chrom, start) and set header SO:coordinate.
                If False (default), set header SO:unsorted.

        !!! Example "Streaming write SAM"
            ```python
            import polars_bio as pb
            lf = pb.scan_bam("input.bam").filter(pl.col("mapping_quality") > 20)
            pb.sink_sam(lf, "filtered.sam")
            ```
        """
        _write_bam_file(lf, path, OutputFormat.Sam, None, sort_on_write=sort_on_write)

    @staticmethod
    def write_cram(
        df: Union[pl.DataFrame, pl.LazyFrame],
        path: str,
        reference_path: str,
        sort_on_write: bool = False,
    ) -> int:
        """
        Write a DataFrame to CRAM format.

        CRAM uses reference-based compression, storing only differences from the
        reference sequence. This achieves 30-60% better compression than BAM.

        Parameters:
            df: DataFrame or LazyFrame with 11 core BAM columns + optional tag columns
            path: Output CRAM file path
            reference_path: Path to reference FASTA file (required). The reference must
                contain all sequences referenced by the alignment data.
            sort_on_write: If True, sort records by (chrom, start) and set header SO:coordinate.
                If False (default), set header SO:unsorted.

        Returns:
            Number of rows written

        !!! warning "Known Limitation: MD and NM Tags"
            Due to a limitation in the underlying noodles-cram library, **MD and NM tags cannot be read back from CRAM files** after writing, even though they are written to the file. If you need MD/NM tags for downstream analysis, use BAM format instead. Other optional tags (RG, MQ, AM, OQ, AS, etc.) work correctly. See: https://github.com/biodatageeks/datafusion-bio-formats/issues/54

        !!! Example "Write CRAM files"
            ```python
            import polars_bio as pb

            df = pb.read_bam("input.bam", tag_fields=["NM", "AS"])

            # Write CRAM with reference (required)
            pb.write_cram(df, "output.cram", reference_path="reference.fasta")

            # For sorted output
            pb.write_cram(df, "output.cram", reference_path="reference.fasta", sort_on_write=True)
            ```
        """
        return _write_bam_file(
            df, path, OutputFormat.Cram, reference_path, sort_on_write=sort_on_write
        )

    @staticmethod
    def sink_cram(
        lf: pl.LazyFrame,
        path: str,
        reference_path: str,
        sort_on_write: bool = False,
    ) -> None:
        """
        Streaming write a LazyFrame to CRAM format.

        CRAM uses reference-based compression, storing only differences from the
        reference sequence. This method streams data without materializing all
        rows in memory.

        Parameters:
            lf: LazyFrame to write
            path: Output CRAM file path
            reference_path: Path to reference FASTA file (required). The reference must
                contain all sequences referenced by the alignment data.
            sort_on_write: If True, sort records by (chrom, start) and set header SO:coordinate.
                If False (default), set header SO:unsorted.

        !!! warning "Known Limitation: MD and NM Tags"
            Due to a limitation in the underlying noodles-cram library, **MD and NM tags cannot be read back from CRAM files** after writing, even though they are written to the file. If you need MD/NM tags for downstream analysis, use BAM format instead. Other optional tags (RG, MQ, AM, OQ, AS, etc.) work correctly. See: https://github.com/biodatageeks/datafusion-bio-formats/issues/54

        !!! Example "Streaming write CRAM"
            ```python
            import polars_bio as pb
            import polars as pl

            lf = pb.scan_bam("large_input.bam")
            lf = lf.filter(pl.col("mapping_quality") > 30)

            # Write CRAM with reference (required)
            pb.sink_cram(lf, "filtered.cram", reference_path="reference.fasta")

            # For sorted output
            pb.sink_cram(lf, "filtered.cram", reference_path="reference.fasta", sort_on_write=True)
            ```
        """
        _write_bam_file(
            lf, path, OutputFormat.Cram, reference_path, sort_on_write=sort_on_write
        )