feat: Add coordinate_system_zero_based parameter to all table providers

This commit implements the coordinate system parameter for all bio format
table providers (VCF, GFF, BAM, BED, CRAM) to support both 0-based and
1-based coordinate output.

Changes:
- Add `coordinate_system_zero_based: bool` parameter to all TableProvider
  constructors
- Store coordinate system preference in Arrow schema metadata with key
  `bio.coordinate_system_zero_based`
- Add `COORDINATE_SYSTEM_METADATA_KEY` constant in bio-format-core
- Update position conversion in physical_exec.rs for each format:
  - When true (default): subtract 1 from noodles 1-based positions
  - When false: use noodles positions as-is (1-based)
- Update all binaries, examples, and tests to pass the new parameter
- Fix test assertions to expect 0-based coordinates (default)

Breaking Change: All TableProvider::new() constructors now require
an additional bool parameter as the last argument.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: mwiewior

.github/workflows/benchmark.yml.bak

@@ -1,8 +1,8 @@
 use anyhow::{Context, Result};
 use datafusion::prelude::*;
 use datafusion_bio_benchmarks_common::{
-    extract_drive_id, write_result, BenchmarkCategory, BenchmarkResultBuilder, DataDownloader,
-    TestDataFile,
+    BenchmarkCategory, BenchmarkResultBuilder, DataDownloader, TestDataFile, extract_drive_id,
+    write_result,
 };
 use datafusion_bio_format_core::object_storage::ObjectStorageOptions;
 use serde::Deserialize;
@@ -210,16 +210,22 @@ async fn register_table(
         "gff" => {
             let storage_options = ObjectStorageOptions::default();
             use datafusion_bio_format_gff::table_provider::GffTableProvider;
-            let provider =
-                GffTableProvider::new(file_path.to_string(), None, None, Some(storage_options))
-                    .context("Failed to create GFF table provider")?;
+            let provider = GffTableProvider::new(
+                file_path.to_string(),
+                None,
+                None,
+                Some(storage_options),
+                true,
+            )
+            .context("Failed to create GFF table provider")?;
             ctx.register_table(table_name, std::sync::Arc::new(provider))
                 .context("Failed to register GFF table")?;
         }
         "vcf" => {
             use datafusion_bio_format_vcf::table_provider::VcfTableProvider;
-            let provider = VcfTableProvider::new(file_path.to_string(), None, None, None, None)
-                .context("Failed to create VCF table provider")?;
+            let provider =
+                VcfTableProvider::new(file_path.to_string(), None, None, None, None, true)
+                    .context("Failed to create VCF table provider")?;
             ctx.register_table(table_name, std::sync::Arc::new(provider))
                 .context("Failed to register VCF table")?;
         }
@@ -232,7 +238,7 @@ async fn register_table(
         }
         "bam" => {
             use datafusion_bio_format_bam::table_provider::BamTableProvider;
-            let provider = BamTableProvider::new(file_path.to_string(), None, None)
+            let provider = BamTableProvider::new(file_path.to_string(), None, None, true)
                 .context("Failed to create BAM table provider")?;
             ctx.register_table(table_name, std::sync::Arc::new(provider))
                 .context("Failed to register BAM table")?;
@@ -241,7 +247,7 @@ async fn register_table(
             use datafusion_bio_format_bed::table_provider::{BEDFields, BedTableProvider};
             // Default to BED3 format (chrom, start, end)
             let provider =
-                BedTableProvider::new(file_path.to_string(), BEDFields::BED3, None, None)
+                BedTableProvider::new(file_path.to_string(), BEDFields::BED3, None, None, true)
                     .context("Failed to create BED table provider")?;
             ctx.register_table(table_name, std::sync::Arc::new(provider))
                 .context("Failed to register BED table")?;

benchmarks/runner/src/main.rs

@@ -14,8 +14,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         concurrent_fetches: Some(1), // Number of concurrent requests
         compression_type: None,
     };
-    let table =
-        BamTableProvider::new(file_path.clone(), Some(1), Some(object_storage_options)).unwrap();
+    let table = BamTableProvider::new(
+        file_path.clone(),
+        Some(1),
+        Some(object_storage_options),
+        true,
+    )
+    .unwrap();
 
     let ctx = datafusion::execution::context::SessionContext::new();
     ctx.sql("set datafusion.execution.skip_physical_aggregate_schema_check=true")

datafusion/bio-format-bam/examples/test_bam_reader.rs

@@ -22,7 +22,7 @@
 //! let ctx = SessionContext::new();
 //!
 //! // Register a BAM file as a table
-//! let table = BamTableProvider::new("data/alignments.bam".to_string(), None, None)?;
+//! let table = BamTableProvider::new("data/alignments.bam".to_string(), None, None, true)?;
 //! ctx.register_table("alignments", Arc::new(table))?;
 //!
 //! // Query with SQL

datafusion/bio-format-bam/src/lib.rs

@@ -30,6 +30,8 @@ pub struct BamExec {
     pub(crate) limit: Option<usize>,
     pub(crate) thread_num: Option<usize>,
     pub(crate) object_storage_options: Option<ObjectStorageOptions>,
+    /// If true, output 0-based half-open coordinates; if false, 1-based closed coordinates
+    pub(crate) coordinate_system_zero_based: bool,
 }
 
 impl Debug for BamExec {
@@ -84,6 +86,7 @@ impl ExecutionPlan for BamExec {
             self.thread_num,
             self.projection.clone(),
             self.object_storage_options.clone(),
+            self.coordinate_system_zero_based,
         );
         let stream = futures::stream::once(fut).try_flatten();
         Ok(Box::pin(RecordBatchStreamAdapter::new(schema, stream)))
@@ -95,6 +98,7 @@ async fn get_remote_bam_stream(
     batch_size: usize,
     projection: Option<Vec<usize>>,
     object_storage_options: Option<ObjectStorageOptions>,
+    coordinate_system_zero_based: bool,
 ) -> datafusion::error::Result<
     AsyncStream<datafusion::error::Result<RecordBatch>, impl Future<Output = ()> + Sized>,
 > {
@@ -140,7 +144,9 @@ async fn get_remote_bam_stream(
             chrom.push(chrom_name);
             match record.alignment_start() {
                 Some(start_pos) => {
-                    start.push(Some(start_pos?.get() as u32));
+                    // noodles normalizes all positions to 1-based; subtract 1 for 0-based output
+                    let pos = start_pos?.get() as u32;
+                    start.push(Some(if coordinate_system_zero_based { pos - 1 } else { pos }));
                 },
                 None => {
                     start.push(None);
@@ -180,7 +186,11 @@ async fn get_remote_bam_stream(
             );
             mate_chrom.push(chrom_name);
             match record.mate_alignment_start()  {
-                Some(start) => mate_start.push(Some(start?.get() as u32)),
+                Some(mate_start_pos) => {
+                    // noodles normalizes all positions to 1-based; subtract 1 for 0-based output
+                    let pos = mate_start_pos?.get() as u32;
+                    mate_start.push(Some(if coordinate_system_zero_based { pos - 1 } else { pos }));
+                },
                 _ => mate_start.push(None),
             };
 
@@ -252,6 +262,7 @@ async fn get_local_bam(
     batch_size: usize,
     thread_num: Option<usize>,
     projection: Option<Vec<usize>>,
+    coordinate_system_zero_based: bool,
 ) -> datafusion::error::Result<impl futures::Stream<Item = datafusion::error::Result<RecordBatch>>>
 {
     let mut name: Vec<Option<String>> = Vec::with_capacity(batch_size);
@@ -298,7 +309,9 @@ async fn get_local_bam(
             };
             match record.alignment_start() {
                 Some(start_pos) => {
-                    start.push(Some(start_pos?.get() as u32));
+                    // noodles normalizes all positions to 1-based; subtract 1 for 0-based output
+                    let pos = start_pos?.get() as u32;
+                    start.push(Some(if coordinate_system_zero_based { pos - 1 } else { pos }));
                 },
                 None => {
                     start.push(None);
@@ -335,7 +348,11 @@ async fn get_local_bam(
             );
             mate_chrom.push(chrom_name);
             match record.mate_alignment_start()  {
-                Some(start) => mate_start.push(Some(start?.get() as u32)),
+                Some(mate_start_pos) => {
+                    // noodles normalizes all positions to 1-based; subtract 1 for 0-based output
+                    let pos = mate_start_pos?.get() as u32;
+                    mate_start.push(Some(if coordinate_system_zero_based { pos - 1 } else { pos }));
+                },
                 None => mate_start.push(None),
             };
 
@@ -484,6 +501,7 @@ async fn get_stream(
     thread_num: Option<usize>,
     projection: Option<Vec<usize>>,
     object_storage_options: Option<ObjectStorageOptions>,
+    coordinate_system_zero_based: bool,
 ) -> datafusion::error::Result<SendableRecordBatchStream> {
     // Open the BGZF-indexed VCF using IndexedReader.
 
@@ -499,6 +517,7 @@ async fn get_stream(
                 batch_size,
                 thread_num,
                 projection,
+                coordinate_system_zero_based,
             )
             .await?;
             Ok(Box::pin(RecordBatchStreamAdapter::new(schema_ref, stream)))
@@ -510,6 +529,7 @@ async fn get_stream(
                 batch_size,
                 projection,
                 object_storage_options,
+                coordinate_system_zero_based,
             )
             .await?;
             Ok(Box::pin(RecordBatchStreamAdapter::new(schema_ref, stream)))

datafusion/bio-format-bam/src/physical_exec.rs

@@ -9,12 +9,14 @@ use datafusion::physical_plan::{
     ExecutionPlan, PlanProperties,
     execution_plan::{Boundedness, EmissionType},
 };
+use datafusion_bio_format_core::COORDINATE_SYSTEM_METADATA_KEY;
 use datafusion_bio_format_core::object_storage::ObjectStorageOptions;
 use log::debug;
 use std::any::Any;
+use std::collections::HashMap;
 use std::sync::Arc;
 
-fn determine_schema() -> datafusion::common::Result<SchemaRef> {
+fn determine_schema(coordinate_system_zero_based: bool) -> datafusion::common::Result<SchemaRef> {
     let fields = vec![
         Field::new("name", DataType::Utf8, true),
         Field::new("chrom", DataType::Utf8, true),
@@ -28,7 +30,13 @@ fn determine_schema() -> datafusion::common::Result<SchemaRef> {
         Field::new("sequence", DataType::Utf8, false),
         Field::new("quality_scores", DataType::Utf8, false),
     ];
-    let schema = Schema::new(fields);
+    // Add coordinate system metadata to schema
+    let mut metadata = HashMap::new();
+    metadata.insert(
+        COORDINATE_SYSTEM_METADATA_KEY.to_string(),
+        coordinate_system_zero_based.to_string(),
+    );
+    let schema = Schema::new_with_metadata(fields, metadata);
     debug!("Schema: {:?}", schema);
     Ok(Arc::new(schema))
 }
@@ -48,6 +56,8 @@ pub struct BamTableProvider {
     thread_num: Option<usize>,
     /// Configuration for cloud storage access
     object_storage_options: Option<ObjectStorageOptions>,
+    /// If true, output 0-based half-open coordinates; if false, 1-based closed coordinates
+    coordinate_system_zero_based: bool,
 }
 
 impl BamTableProvider {
@@ -58,6 +68,8 @@ impl BamTableProvider {
     /// * `file_path` - Path to the BAM file (local or remote URL)
     /// * `thread_num` - Optional number of threads for BGZF decompression
     /// * `object_storage_options` - Optional configuration for cloud storage access
+    /// * `coordinate_system_zero_based` - If true (default), output 0-based half-open coordinates;
+    ///   if false, output 1-based closed coordinates
     ///
     /// # Returns
     ///
@@ -73,6 +85,7 @@ impl BamTableProvider {
     ///     "data/alignments.bam".to_string(),
     ///     Some(4),  // Use 4 threads
     ///     None,     // No cloud storage
+    ///     true,     // Use 0-based coordinates (default)
     /// )?;
     /// # Ok(())
     /// # }
@@ -81,13 +94,15 @@ impl BamTableProvider {
         file_path: String,
         thread_num: Option<usize>,
         object_storage_options: Option<ObjectStorageOptions>,
+        coordinate_system_zero_based: bool,
     ) -> datafusion::common::Result<Self> {
-        let schema = determine_schema()?;
+        let schema = determine_schema(coordinate_system_zero_based)?;
         Ok(Self {
             file_path,
             schema,
             thread_num,
             object_storage_options,
+            coordinate_system_zero_based,
         })
     }
 }
@@ -120,12 +135,19 @@ impl TableProvider for BamTableProvider {
         fn project_schema(schema: &SchemaRef, projection: Option<&Vec<usize>>) -> SchemaRef {
             match projection {
                 Some(indices) if indices.is_empty() => {
-                    Arc::new(Schema::new(vec![Field::new("dummy", DataType::Null, true)]))
+                    // For empty projections (COUNT(*)), use a dummy field with preserved metadata
+                    Arc::new(Schema::new_with_metadata(
+                        vec![Field::new("dummy", DataType::Null, true)],
+                        schema.metadata().clone(),
+                    ))
                 }
                 Some(indices) => {
                     let projected_fields: Vec<Field> =
                         indices.iter().map(|&i| schema.field(i).clone()).collect();
-                    Arc::new(Schema::new(projected_fields))
+                    Arc::new(Schema::new_with_metadata(
+                        projected_fields,
+                        schema.metadata().clone(),
+                    ))
                 }
                 None => schema.clone(),
             }
@@ -146,6 +168,7 @@ impl TableProvider for BamTableProvider {
             limit,
             thread_num: self.thread_num,
             object_storage_options: self.object_storage_options.clone(),
+            coordinate_system_zero_based: self.coordinate_system_zero_based,
         }))
     }
 }

datafusion/bio-format-bam/src/table_provider.rs

@@ -32,6 +32,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         BEDFields::BED4,
         Some(1),
         Some(object_storage_options),
+        true, // Use 0-based coordinates (default)
     )
     .unwrap();
     let ctx = datafusion::execution::context::SessionContext::new();

datafusion/bio-format-bed/examples/test_bed_reader.rs

@@ -21,7 +21,7 @@
 //! let ctx = SessionContext::new();
 //!
 //! // Register a BED file as a table
-//! let table = BedTableProvider::new("data/genes.bed".to_string(), BEDFields::BED3, None, None)?;
+//! let table = BedTableProvider::new("data/genes.bed".to_string(), BEDFields::BED3, None, None, true)?;
 //! ctx.register_table("genes", Arc::new(table))?;
 //!
 //! // Query with SQL