Implement DataFusion FileFormat trait for STORED AS zarr

[jayendra13]

Summary

zarr-datafusion currently integrates with DataFusion via a custom TableProvider (ZarrTable) and TableProviderFactory (ZarrTableFactory). While this works, implementing DataFusion's FileFormat trait would enable richer integration: URL table support, STORED AS zarr DDL syntax, automatic file pruning, and better alignment with DataFusion's execution model.

Current Integration

// datasource/factory.rs — manual TableProviderFactory
impl TableProviderFactory for ZarrTableFactory {
    async fn create(&self, _state: &dyn Session, cmd: &CreateExternalTable) -> Result<Arc<dyn TableProvider>> {
        let table = ZarrTable::try_new(&cmd.location).await?;
        Ok(Arc::new(table))
    }
}

// datasource/zarr.rs — custom TableProvider
impl TableProvider for ZarrTable {
    fn supports_filters_pushdown(&self, filters: &[&Expr]) -> Result<Vec<TableProviderFilterPushDown>> {
        Ok(filters.iter().map(|_| TableProviderFilterPushDown::Inexact).collect())
    }
    // ...
}

What's missing vs. FileFormat-based approach:

Feature	Current (TableProvider)	With FileFormat
CREATE EXTERNAL TABLE ... STORED AS zarr	Works via factory	Works natively
URL tables (SELECT * FROM 'gs://bucket/data.zarr')	Not supported	Automatic
File pruning via statistics	Not supported	Built-in
Partition discovery	Not supported	Via ListingTable
Write support (INSERT INTO)	Not supported	Via DataSink
Schema inference from path	Manual in factory	Via infer_schema() trait method
Statistics for optimizer	Partial (coord bounds only)	Full column-level stats

Proposed Change

Implement FileFormat and FileFormatFactory traits:

pub struct ZarrFormat {
    // Options like default version detection, stat caching, etc.
}

#[async_trait]
impl FileFormat for ZarrFormat {
    fn as_any(&self) -> &dyn Any { self }

    async fn infer_schema(&self, state: &SessionState, store: &Arc<dyn ObjectStore>, objects: &[ObjectMeta]) -> Result<SchemaRef> {
        // Use existing schema_inference::infer_schema()
    }

    async fn infer_stats(&self, state: &SessionState, store: &Arc<dyn ObjectStore>, table_schema: SchemaRef, object: &ObjectMeta) -> Result<Statistics> {
        // Return coord bounds as column-level min/max
    }

    async fn create_physical_plan(&self, state: &SessionState, conf: FileScanConfig) -> Result<Arc<dyn ExecutionPlan>> {
        // Create ZarrExec with projection, filter, limit from conf
    }

    fn file_type(&self) -> FileType {
        FileType::new("ZARR")  // or register extension
    }
}

pub struct ZarrFormatFactory;

impl FileFormatFactory for ZarrFormatFactory {
    fn create(&self, _state: &SessionState, _options: &HashMap<String, String>) -> Result<Arc<dyn FileFormat>> {
        Ok(Arc::new(ZarrFormat::default()))
    }
}

Registration:

let factory = Arc::new(ZarrFormatFactory);
let state = SessionStateBuilder::new()
    .with_default_features()
    .with_file_formats(vec![factory])
    .build();
let ctx = SessionContext::new_with_state(state).enable_url_table();

// Now works:
ctx.sql("SELECT * FROM 'gs://bucket/era5.zarr' WHERE time = '2024-01-01'").await?;
ctx.sql("CREATE EXTERNAL TABLE era5 STORED AS zarr LOCATION 's3://bucket/era5.zarr'").await?;

Considerations

• Zarr stores are directories, not files — FileFormat assumes file-level granularity. We'd treat each .zarr directory as a single "file" and handle internal chunk/array discovery within the format.
• Keep existing TableProvider — The ZarrTable API is useful for programmatic registration. FileFormat adds SQL-level integration on top.
• Backward compatibility — Existing ZarrTableFactory continues to work; FileFormat is additive.

Files to Modify

• src/datasource/format.rs (new) — ZarrFormat implementing FileFormat, ZarrFormatFactory implementing FileFormatFactory
• src/datasource/mod.rs — Export new format module
• src/lib.rs — Re-export for library users
• src/bin/zarr_cli/main.rs — Register ZarrFormatFactory in CLI session

Motivation

Inspired by Vortex's VortexFormat / VortexFormatFactory DataFusion integration, which provides URL table support, file pruning, metadata caching, and expression conversion as first-class features through the FileFormat trait. This is the idiomatic way to integrate custom file formats with DataFusion.