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

@@ -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

@@ -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,
         })
     }
 }
@@ -146,6 +161,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

@@ -41,6 +41,8 @@ pub struct BedExec {
     pub(crate) thread_num: Option<usize>,
     /// Optional cloud storage configuration
     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 BedExec {
@@ -95,7 +97,7 @@ impl ExecutionPlan for BedExec {
         _partition: usize,
         context: Arc<TaskContext>,
     ) -> datafusion::common::Result<SendableRecordBatchStream> {
-        debug!("GffExec::execute");
+        debug!("BedExec::execute");
         debug!("Projection: {:?}", self.projection);
         let batch_size = context.session_config().batch_size();
         let schema = self.schema.clone();
@@ -107,6 +109,7 @@ impl ExecutionPlan for BedExec {
             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)))
@@ -123,13 +126,15 @@ impl ExecutionPlan for BedExec {
 /// * `batch_size` - Number of records per batch
 /// * `projection` - Optional column projection
 /// * `object_storage_options` - Cloud storage configuration
+/// * `coordinate_system_zero_based` - If true, output 0-based coordinates; if false, 1-based
 async fn get_remote_bed_stream(
     file_path: String,
     bed_fields: BEDFields,
     schema: SchemaRef,
     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>,
 > {
@@ -158,8 +163,16 @@ async fn get_remote_bed_stream(
         while let Some(result) = records.next().await {
             let record = result?;  // propagate errors if any
             chroms.push(record.reference_sequence_name().to_string());
-            poss.push(record.feature_start()?.get() as u32);
-            pose.push(record.feature_end().unwrap()?.get() as u32);
+            // BED files are natively 0-based in noodles. feature_start().get() returns 1-based.
+            // For 0-based output: subtract 1 to get back to 0-based
+            // For 1-based output: use as-is (noodles already returns 1-based)
+            let start_pos = record.feature_start()?.get() as u32;
+            poss.push(if coordinate_system_zero_based { start_pos - 1 } else { start_pos });
+            // End position: noodles returns 1-based exclusive end
+            // For 0-based half-open: subtract 1 to get 0-based exclusive end
+            // For 1-based closed: use as-is
+            let end_pos = record.feature_end().unwrap()?.get() as u32;
+            pose.push(if coordinate_system_zero_based { end_pos - 1 } else { end_pos });
             name.push(record.name().map(|n| n.to_string()));
 
             record_num += 1;
@@ -195,7 +208,6 @@ async fn get_remote_bed_stream(
                 &pose,
                 &name,
                 projection.clone(),
-                // if infos.is_empty() { None } else { Some(&infos) },
             )?;
             yield batch;
         }
@@ -213,13 +225,15 @@ async fn get_remote_bed_stream(
 /// * `batch_size` - Number of records per batch
 /// * `thread_num` - Number of threads for parallel BGZF decompression
 /// * `projection` - Optional column projection
+/// * `coordinate_system_zero_based` - If true, output 0-based coordinates; if false, 1-based
 async fn get_local_bed(
     file_path: String,
     bed_fields: BEDFields,
     schema: SchemaRef,
     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 reader = match bed_fields {
@@ -244,8 +258,16 @@ async fn get_local_bed(
         while let Some(result) = records.next().await {
             let record = result?;  // propagate errors if any
             chroms.push(record.reference_sequence_name().to_string());
-            poss.push(record.feature_start()?.get() as u32);
-            pose.push(record.feature_end().unwrap()?.get() as u32);
+            // BED files are natively 0-based in noodles. feature_start().get() returns 1-based.
+            // For 0-based output: subtract 1 to get back to 0-based
+            // For 1-based output: use as-is (noodles already returns 1-based)
+            let start_pos = record.feature_start()?.get() as u32;
+            poss.push(if coordinate_system_zero_based { start_pos - 1 } else { start_pos });
+            // End position: noodles returns 1-based exclusive end
+            // For 0-based half-open: subtract 1 to get 0-based exclusive end
+            // For 1-based closed: use as-is
+            let end_pos = record.feature_end().unwrap()?.get() as u32;
+            pose.push(if coordinate_system_zero_based { end_pos - 1 } else { end_pos });
             name.push(record.name().map(|n| n.to_string()));
 
             record_num += 1;
@@ -279,8 +301,8 @@ async fn get_local_bed(
                 &chroms,
                 &poss,
                 &pose,
-                &name
-                , projection.clone(),
+                &name,
+                projection.clone(),
             )?;
             yield batch;
         }
@@ -351,6 +373,7 @@ fn build_record_batch(
 /// * `thread_num` - Optional thread count for parallel reading
 /// * `projection` - Optional column projection
 /// * `object_storage_options` - Cloud storage configuration
+/// * `coordinate_system_zero_based` - If true, output 0-based coordinates; if false, 1-based
 async fn get_stream(
     file_path: String,
     bed_fields: BEDFields,
@@ -359,8 +382,9 @@ 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.
+    // Open the BGZF-indexed BED file.
 
     let file_path = file_path.clone();
     let store_type = get_storage_type(file_path.clone());
@@ -375,6 +399,7 @@ async fn get_stream(
                 batch_size,
                 thread_num,
                 projection,
+                coordinate_system_zero_based,
             )
             .await?;
             Ok(Box::pin(RecordBatchStreamAdapter::new(schema_ref, stream)))
@@ -387,6 +412,7 @@ async fn get_stream(
                 batch_size,
                 projection,
                 object_storage_options,
+                coordinate_system_zero_based,
             )
             .await?;
             Ok(Box::pin(RecordBatchStreamAdapter::new(schema_ref, stream)))