From 241e983f23e4d316a64c7f8319b846304d0a6243 Mon Sep 17 00:00:00 2001 From: Vlad Pankratov Date: Wed, 6 May 2026 13:54:54 +0200 Subject: [PATCH 1/7] Initial commit --- .cargo/config.toml | 15 +++ .github/workflows/ci-rust.yml | 79 +++++++++++++ .gitignore | 4 + rust/Cargo.toml | 29 +++++ rust/crates/odc-sys/Cargo.toml | 32 +++++ rust/crates/odc-sys/README.md | 18 +++ rust/crates/odc-sys/build.rs | 156 +++++++++++++++++++++++++ rust/crates/odc-sys/cpp/odc_bridge.cpp | 144 +++++++++++++++++++++++ rust/crates/odc-sys/cpp/odc_bridge.h | 122 +++++++++++++++++++ rust/crates/odc-sys/src/lib.rs | 130 +++++++++++++++++++++ 10 files changed, 729 insertions(+) create mode 100644 .cargo/config.toml create mode 100644 .github/workflows/ci-rust.yml create mode 100644 rust/Cargo.toml create mode 100644 rust/crates/odc-sys/Cargo.toml create mode 100644 rust/crates/odc-sys/README.md create mode 100644 rust/crates/odc-sys/build.rs create mode 100644 rust/crates/odc-sys/cpp/odc_bridge.cpp create mode 100644 rust/crates/odc-sys/cpp/odc_bridge.h create mode 100644 rust/crates/odc-sys/src/lib.rs diff --git a/.cargo/config.toml b/.cargo/config.toml new file mode 100644 index 00000000..b4069aba --- /dev/null +++ b/.cargo/config.toml @@ -0,0 +1,15 @@ +[build] +jobs = -1 + +[target.'cfg(all())'] +rustflags = [ + "-Wclippy::all", + "-Wclippy::pedantic", + "-Wclippy::nursery", + "-Wclippy::unwrap_used", + "-Aclippy::module_name_repetitions", + "-Aclippy::missing_errors_doc", +] + +[net] +git-fetch-with-cli = true diff --git a/.github/workflows/ci-rust.yml b/.github/workflows/ci-rust.yml new file mode 100644 index 00000000..48efac09 --- /dev/null +++ b/.github/workflows/ci-rust.yml @@ -0,0 +1,79 @@ +name: rust + +on: + push: + branches: + - 'master' + - 'develop' + - 'rust-bindings' + tags-ignore: + - '**' + paths: + - 'rust/**' + - '.github/workflows/ci-rust.yml' + + pull_request: + paths: + - 'rust/**' + - '.github/workflows/ci-rust.yml' + + workflow_dispatch: ~ + +env: + CARGO_TERM_COLOR: always + CARGO_NET_GIT_FETCH_WITH_CLI: "true" + +jobs: + fmt: + name: fmt + runs-on: ubuntu-latest + defaults: + run: + working-directory: rust + steps: + - uses: actions/checkout@v4 + + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt + + - name: Format check + run: cargo fmt --check + + clippy: + name: clippy + if: ${{ !github.event.pull_request.head.repo.fork }} + runs-on: ubuntu-latest + defaults: + run: + working-directory: rust + steps: + - uses: actions/checkout@v4 + + - name: Configure git for private repos + run: git config --global url."https://x-access-token:${{ secrets.GH_REPO_READ_TOKEN }}@github.com/".insteadOf "ssh://git@github.com/" + + - uses: dtolnay/rust-toolchain@stable + with: + components: clippy + + - name: Clippy + run: cargo clippy --features vendored --all-targets -- -D warnings + + test: + name: test + if: ${{ !github.event.pull_request.head.repo.fork }} + runs-on: ubuntu-latest + defaults: + run: + working-directory: rust + steps: + - uses: actions/checkout@v4 + + - name: Configure git for private repos + run: git config --global url."https://x-access-token:${{ secrets.GH_REPO_READ_TOKEN }}@github.com/".insteadOf "ssh://git@github.com/" + + - uses: dtolnay/rust-toolchain@stable + + - name: Test + run: cargo test --features vendored diff --git a/.gitignore b/.gitignore index 1fed26d0..6701a760 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,7 @@ Testing/* tests/core/Testing/* build docs/_build + +# Rust +rust/target/ +rust/Cargo.lock diff --git a/rust/Cargo.toml b/rust/Cargo.toml new file mode 100644 index 00000000..e1f62f4c --- /dev/null +++ b/rust/Cargo.toml @@ -0,0 +1,29 @@ +[workspace] +resolver = "2" +members = ["crates/odc-sys"] + +[workspace.package] +edition = "2024" +license = "Apache-2.0" +repository = "https://github.com/ecmwf/odc" +rust-version = "1.90" +readme = "README.md" +keywords = ["ecmwf", "weather", "meteorology", "odb"] +categories = ["science"] + +[workspace.dependencies] +# Internal +odc-sys = { path = "crates/odc-sys" } + +# Foundation crates +eckit-sys = { git = "ssh://git@github.com/ecmwf/eckit.git", branch = "rust-bindings", default-features = false } + +# Build tools +bindman = { git = "ssh://git@github.com/ecmwf/bindman.git", branch = "generate_exception_bridge" } +bindman-build = { git = "ssh://git@github.com/ecmwf/bindman.git", branch = "generate_exception_bridge" } +bindman-utils = { git = "ssh://git@github.com/ecmwf/bindman.git", branch = "generate_exception_bridge" } + +# External +cxx = "1.0" +cxx-build = "1.0" +thiserror = "2" diff --git a/rust/crates/odc-sys/Cargo.toml b/rust/crates/odc-sys/Cargo.toml new file mode 100644 index 00000000..ae57f622 --- /dev/null +++ b/rust/crates/odc-sys/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "odc-sys" +version = "1.6.3" +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +readme.workspace = true +keywords.workspace = true +categories.workspace = true +description = "C++ bindings to ECMWF odc (ODB-2 encoder/decoder) library using cxx" +links = "odc_sys" +build = "build.rs" + +[features] +default = ["vendored"] + +# Build strategy (mutually exclusive) +vendored = ["eckit-sys/vendored"] +system = ["eckit-sys/system"] + +[dependencies] +cxx.workspace = true +eckit-sys = { workspace = true, default-features = false, features = ["eckit-sql"] } +bindman.workspace = true + +[build-dependencies] +cxx-build.workspace = true +bindman-utils.workspace = true +bindman-build.workspace = true + +[package.metadata.docs.rs] diff --git a/rust/crates/odc-sys/README.md b/rust/crates/odc-sys/README.md new file mode 100644 index 00000000..bd256af4 --- /dev/null +++ b/rust/crates/odc-sys/README.md @@ -0,0 +1,18 @@ +# odc-sys + +Low-level Rust bindings to ECMWF's [odc](https://github.com/ecmwf/odc) (ODB-2 encoder/decoder) C++ library. + +This crate provides raw FFI bindings using [cxx](https://cxx.rs/). For a safe, ergonomic API, use the higher-level `odc` crate (planned). + +## Features + +### Build strategy (mutually exclusive) + +- `vendored` - Build odc and its dependencies (eckit) from source. +- `system` - Link against system-installed odc. + +`vendored` is enabled by default. + +## License + +Apache-2.0 diff --git a/rust/crates/odc-sys/build.rs b/rust/crates/odc-sys/build.rs new file mode 100644 index 00000000..48fddcd2 --- /dev/null +++ b/rust/crates/odc-sys/build.rs @@ -0,0 +1,156 @@ +use std::env; +use std::path::PathBuf; + +const ODC_VERSION: &str = "1.6.3"; + +fn main() { + println!("cargo:rerun-if-changed=build.rs"); + println!("cargo:rerun-if-changed=src/lib.rs"); + println!("cargo:rerun-if-changed=cpp/odc_bridge.h"); + println!("cargo:rerun-if-changed=cpp/odc_bridge.cpp"); + println!("cargo:rerun-if-env-changed=ODC_DIR"); + println!("cargo:rerun-if-env-changed=DOCS_RS"); + + if bindman_utils::is_docs_rs() { + return; + } + + bindman_utils::validate_build_mode(cfg!(feature = "system"), cfg!(feature = "vendored")); + + if cfg!(feature = "system") { + build_system(); + } else { + build_vendored(); + } +} + +#[cfg(feature = "system")] +fn build_system() { + let crate_dir = + PathBuf::from(env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set")); + + let eckit_include = env::var("DEP_ECKIT_SYS_INCLUDE").expect("DEP_ECKIT_SYS_INCLUDE not set"); + let eckit_out_dir = env::var("DEP_ECKIT_SYS_OUT_DIR").expect("DEP_ECKIT_SYS_OUT_DIR not set"); + let eckit_cpp_dir = env::var("DEP_ECKIT_SYS_CPP_DIR").expect("DEP_ECKIT_SYS_CPP_DIR not set"); + + let (root, odc_include, lib_dir) = bindman_utils::cmake_find_package("odc", ODC_VERSION); + + println!("cargo:rustc-link-search=native={}", lib_dir.display()); + println!("cargo:rustc-link-lib=dylib=odccore"); + + cxx_build::bridge("src/lib.rs") + .file(crate_dir.join("cpp/odc_bridge.cpp")) + .include(&odc_include) + .include(&eckit_include) + .include(&eckit_out_dir) + .include(&eckit_cpp_dir) + .include(crate_dir.join("cpp")) + .flag_if_supported("-std=c++17") + .compile("odc_sys_bridge"); + + bindman_utils::link_cpp_stdlib(); + + println!("cargo:root={}", root.display()); + println!("cargo:include={}", odc_include.display()); + + bindman_build::check_cpp_api(&odc_include, &crate_dir.join("src/lib.rs")); +} + +#[cfg(not(feature = "system"))] +fn build_system() { + unreachable!("build_system called without system feature"); +} + +#[cfg(feature = "vendored")] +fn build_vendored() { + use std::fs; + use std::process::Command; + + const ECBUILD_REPO: &str = "https://github.com/ecmwf/ecbuild.git"; + const ECBUILD_TAG: &str = "3.13.1"; + const ODC_REPO: &str = "https://github.com/ecmwf/odc.git"; + + let out_dir = PathBuf::from(env::var("OUT_DIR").expect("OUT_DIR not set")); + let src_dir = out_dir.join("src"); + let build_dir = out_dir.join("build"); + let install_dir = out_dir.join("install"); + + fs::create_dir_all(&src_dir).expect("Failed to create src directory"); + fs::create_dir_all(&build_dir).expect("Failed to create build directory"); + + let eckit_root = env::var("DEP_ECKIT_SYS_ROOT").expect("DEP_ECKIT_SYS_ROOT not set"); + let eckit_out_dir = env::var("DEP_ECKIT_SYS_OUT_DIR").expect("DEP_ECKIT_SYS_OUT_DIR not set"); + let eckit_cpp_dir = env::var("DEP_ECKIT_SYS_CPP_DIR").expect("DEP_ECKIT_SYS_CPP_DIR not set"); + + let ecbuild_src = bindman_utils::git_clone(ECBUILD_REPO, ECBUILD_TAG, &src_dir.join("ecbuild")); + let odc_src = bindman_utils::git_clone(ODC_REPO, ODC_VERSION, &src_dir.join("odc")); + + let ecbuild_bin = ecbuild_src.join("bin/ecbuild"); + let num_jobs = bindman_utils::build_parallelism(); + + let cmake_prefix_path = eckit_root.clone(); + + let mut cmd = Command::new(&ecbuild_bin); + cmd.current_dir(&build_dir) + .arg(format!("--prefix={}", install_dir.display())) + .arg("--") + .arg(&odc_src) + .arg(format!("-DCMAKE_PREFIX_PATH={cmake_prefix_path}")) + .arg(format!( + "-DCMAKE_BUILD_TYPE={}", + bindman_utils::cmake_build_type() + )) + .arg("-DENABLE_TESTS=OFF") + .arg("-DBUILD_TESTING=OFF") + .arg("-DENABLE_DOCS=OFF") + .arg("-DENABLE_FORTRAN=OFF") + .arg("-DENABLE_PYTHON=OFF"); + + #[cfg(target_os = "macos")] + cmd.arg("-DCMAKE_INSTALL_NAME_DIR=@rpath"); + + bindman_utils::run_command(&mut cmd, "ecbuild configure odc"); + + bindman_utils::run_command( + Command::new("cmake") + .args(["--build", ".", "--parallel", &num_jobs]) + .current_dir(&build_dir), + "cmake build odc", + ); + + bindman_utils::run_command( + Command::new("cmake") + .args(["--install", "."]) + .current_dir(&build_dir), + "cmake install odc", + ); + + let include_dir = install_dir.join("include"); + let crate_dir = + PathBuf::from(env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set")); + let lib_dir = bindman_utils::resolve_lib_dir(&install_dir); + + cxx_build::bridge("src/lib.rs") + .file(crate_dir.join("cpp/odc_bridge.cpp")) + .include(&include_dir) + .include(format!("{eckit_root}/include")) + .include(&eckit_out_dir) + .include(&eckit_cpp_dir) + .include(crate_dir.join("cpp")) + .flag_if_supported("-std=c++17") + .compile("odc_sys_bridge"); + + println!("cargo:rustc-link-search=native={}", lib_dir.display()); + println!("cargo:rustc-link-lib=dylib=odccore"); + bindman_utils::link_cpp_stdlib(); + + println!("cargo:root={}", install_dir.display()); + println!("cargo:include={}", include_dir.display()); + + bindman_build::check_cpp_api(&include_dir, &crate_dir.join("src/lib.rs")); +} + +#[cfg(not(feature = "vendored"))] +fn build_vendored() { + unreachable!("build_vendored called without vendored feature"); +} diff --git a/rust/crates/odc-sys/cpp/odc_bridge.cpp b/rust/crates/odc-sys/cpp/odc_bridge.cpp new file mode 100644 index 00000000..80e22943 --- /dev/null +++ b/rust/crates/odc-sys/cpp/odc_bridge.cpp @@ -0,0 +1,144 @@ +// odc C++ bridge implementation +#include "odc_bridge.h" + +#include "eckit/io/MemoryHandle.h" + +#include + +namespace odc_bridge { + +// ==================== SelectIteratorWrapper ==================== + +SelectIteratorWrapper::SelectIteratorWrapper(odc::Select& select) : current_(select.begin()), end_(select.end()) {} + +SelectIteratorWrapper::SelectIteratorWrapper(odc::Select::iterator current, odc::Select::iterator end) : + current_(std::move(current)), end_(std::move(end)) {} + +bool SelectIteratorWrapper::valid() { + return current_ != end_; +} + +void SelectIteratorWrapper::advance() { + ++current_; +} + +size_t SelectIteratorWrapper::column_count() const { + return current_->columns().size(); +} + +rust::String SelectIteratorWrapper::column_name(size_t idx) const { + return rust::String(current_->columns()[idx]->name()); +} + +ColumnType SelectIteratorWrapper::column_type(size_t idx) const { + return current_->columns()[idx]->type(); +} + +double SelectIteratorWrapper::data(size_t idx) const { + return current_->data(idx); +} + +rust::String SelectIteratorWrapper::data_string(size_t idx) { + return rust::String(current_->string(idx)); +} + +int64_t SelectIteratorWrapper::data_integer(size_t idx) { + return current_->integer(idx); +} + +// ==================== SelectWrapper ==================== + +SelectWrapper::SelectWrapper(rust::Str sql, eckit_bridge::DataHandleWrapper& handle) : + select_(std::make_unique(std::string(sql), handle.inner())) {} + +std::unique_ptr SelectWrapper::begin() { + return std::make_unique(*select_); +} + +std::unique_ptr SelectWrapper::createSelectIterator(rust::Str sql) { + auto* it = select_->createSelectIterator(std::string(sql)); + it->next(); + return std::make_unique(odc::Select::iterator(it), select_->end()); +} + +rust::String SelectWrapper::database_name() { + return rust::String(select_->database().name()); +} + +std::unique_ptr select_create(rust::Str sql, eckit_bridge::DataHandleWrapper& handle) { + return std::make_unique(sql, handle); +} + +// ==================== WriteIteratorWrapper ==================== + +WriteIteratorWrapper::WriteIteratorWrapper(odc::Writer<>::iterator iter) : iter_(std::move(iter)) {} + +void WriteIteratorWrapper::set_column(size_t index, rust::Str name, ColumnType col_type) { + iter_->setColumn(index, std::string(name), col_type); +} + +void WriteIteratorWrapper::set_number_of_columns(size_t n) { + iter_->setNumberOfColumns(n); +} + +void WriteIteratorWrapper::set_data(size_t index, double value) { + iter_->data(index) = value; +} + +void WriteIteratorWrapper::set_data_string(size_t index, rust::Str value) { + size_t maxlen = sizeof(double) * iter_->columns()[index]->dataSizeDoubles(); + ::strncpy(reinterpret_cast(&iter_->data(index)), std::string(value).c_str(), maxlen); +} + +void WriteIteratorWrapper::set_data_integer(size_t index, int64_t value) { + iter_->data(index) = static_cast(value); +} + +void WriteIteratorWrapper::set_missing_value(size_t index, double value) { + iter_->missingValue(index, value); +} + +void WriteIteratorWrapper::write_row() { + ++iter_; +} + +void WriteIteratorWrapper::close() { + iter_->close(); +} + +// ==================== WriterWrapper ==================== + +WriterWrapper::WriterWrapper(eckit_bridge::DataHandleWrapper& handle) : + writer_(std::make_unique>(handle.inner())), outit_(writer_->begin()) {} + +std::unique_ptr WriterWrapper::begin() { + return std::make_unique(writer_->begin()); +} + +void WriterWrapper::pass1(SelectWrapper& select) { + auto it = select.select_->begin(); + auto end = select.select_->end(); + outit_->pass1(it, end); +} + +size_t WriterWrapper::rows_buffer_size() const { + return writer_->rowsBufferSize(); +} + +void WriterWrapper::set_rows_buffer_size(size_t n) { + writer_->rowsBufferSize(n); +} + +// Note: data_handle() not exposed — Writer owns the DataHandle internally +// and it's the same one passed to the constructor. Access it via the +// original DataHandleWrapper on the Rust side. + +rust::String WriterWrapper::path() const { + return rust::String(writer_->path()); +} + +std::unique_ptr writer_create(eckit_bridge::DataHandleWrapper& handle) { + return std::make_unique(handle); +} + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/odc_bridge.h b/rust/crates/odc-sys/cpp/odc_bridge.h new file mode 100644 index 00000000..0b066561 --- /dev/null +++ b/rust/crates/odc-sys/cpp/odc_bridge.h @@ -0,0 +1,122 @@ +// odc C++ bridge for Rust FFI +#pragma once + +#include "eckit_bridge.h" +#include "eckit_exceptions.h" + +#include "odc/Select.h" +#include "odc/Writer.h" +#include "odc/api/ColumnType.h" + +#include "rust/cxx.h" + +#include +#include +#include + +namespace odc_bridge { + +// ColumnType is odc::api::ColumnType — cxx static_asserts values match. +using odc::api::ColumnType; + +// ==================== SelectIteratorWrapper ==================== + +/// Wraps `odc::Select::iterator` pair for row-by-row iteration. +class SelectIteratorWrapper { + odc::Select::iterator current_; + odc::Select::iterator end_; + +public: + + SelectIteratorWrapper(odc::Select& select); + SelectIteratorWrapper(odc::Select::iterator current, odc::Select::iterator end); + + bool valid(); + void advance(); + + size_t column_count() const; + rust::String column_name(size_t idx) const; + ColumnType column_type(size_t idx) const; + double data(size_t idx) const; + rust::String data_string(size_t idx); + int64_t data_integer(size_t idx); +}; + +// ==================== SelectWrapper ==================== + +/// Wraps `odc::Select` — owns the query, produces iterators. +class SelectWrapper { + std::unique_ptr select_; + + friend class WriterWrapper; + +public: + + SelectWrapper(rust::Str sql, eckit_bridge::DataHandleWrapper& handle); + + std::unique_ptr begin(); + std::unique_ptr createSelectIterator(rust::Str sql); + rust::String database_name(); +}; + +std::unique_ptr select_create(rust::Str sql, eckit_bridge::DataHandleWrapper& handle); + +// ==================== WriteIteratorWrapper ==================== + +/// Wraps `odc::Writer<>::iterator` for row-by-row writing. +class WriteIteratorWrapper { + odc::Writer<>::iterator iter_; + +public: + + explicit WriteIteratorWrapper(odc::Writer<>::iterator iter); + + /// Define a column. Must be called before writing any rows. + void set_column(size_t index, rust::Str name, ColumnType col_type); + + /// Set number of columns. + void set_number_of_columns(size_t n); + + /// Set a double value at column index for the current row. + void set_data(size_t index, double value); + + /// Set a string value at column index for the current row. + void set_data_string(size_t index, rust::Str value); + + /// Set an integer value at column index for the current row. + void set_data_integer(size_t index, int64_t value); + + /// Set the missing value for a column. + void set_missing_value(size_t index, double value); + + /// Write the current row (advances the iterator). + void write_row(); + + /// Close the writer. + void close(); +}; + +// ==================== WriterWrapper ==================== + +/// Wraps `odc::Writer<>` — writes filtered ODB data via pass1. +class WriterWrapper { + std::unique_ptr> writer_; + odc::Writer<>::iterator outit_; + +public: + + explicit WriterWrapper(eckit_bridge::DataHandleWrapper& handle); + + void pass1(SelectWrapper& select); + + /// Get a write iterator for row-by-row writing. + std::unique_ptr begin(); + + size_t rows_buffer_size() const; + void set_rows_buffer_size(size_t n); + rust::String path() const; +}; + +std::unique_ptr writer_create(eckit_bridge::DataHandleWrapper& handle); + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/src/lib.rs b/rust/crates/odc-sys/src/lib.rs new file mode 100644 index 00000000..5f301c65 --- /dev/null +++ b/rust/crates/odc-sys/src/lib.rs @@ -0,0 +1,130 @@ +//! FFI bindings to ECMWF odc (ODB-2 encoder/decoder) library. + +use bindman::track_cpp_api; + +#[track_cpp_api( + ("odc/Select.h", class = "Select"), + ("odc/Writer.h", class = "Writer"), + ignore = ["end", "dataHandle"] +)] +#[cxx::bridge(namespace = "odc_bridge")] +pub mod ffi { + /// ODB column data types — compile-time verified against C++ `odc::api::ColumnType`. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + #[repr(i32)] + enum ColumnType { + #[cxx_name = "IGNORE"] + Ignore = 0, + #[cxx_name = "INTEGER"] + Integer = 1, + #[cxx_name = "REAL"] + Real = 2, + #[cxx_name = "STRING"] + String = 3, + #[cxx_name = "BITFIELD"] + Bitfield = 4, + #[cxx_name = "DOUBLE"] + Double = 5, + } + + unsafe extern "C++" { + include!("odc_bridge.h"); + + // Verify ColumnType matches C++ odc::api::ColumnType at compile time + #[namespace = "odc::api"] + type ColumnType; + + // Cross-crate ExternType from eckit-sys + #[namespace = "eckit_bridge"] + type DataHandleWrapper = eckit_sys::DataHandleWrapper; + + // ==================== SelectIteratorWrapper ==================== + + type SelectIteratorWrapper; + + #[must_use] + fn valid(self: Pin<&mut SelectIteratorWrapper>) -> bool; + fn advance(self: Pin<&mut SelectIteratorWrapper>) -> Result<()>; + fn column_count(self: &SelectIteratorWrapper) -> usize; + fn column_name(self: &SelectIteratorWrapper, idx: usize) -> Result; + fn column_type(self: &SelectIteratorWrapper, idx: usize) -> Result; + fn data(self: &SelectIteratorWrapper, idx: usize) -> Result; + fn data_string(self: Pin<&mut SelectIteratorWrapper>, idx: usize) -> Result; + fn data_integer(self: Pin<&mut SelectIteratorWrapper>, idx: usize) -> Result; + + // ==================== SelectWrapper ==================== + + type SelectWrapper; + + fn begin(self: Pin<&mut SelectWrapper>) -> Result>; + #[cxx_name = "createSelectIterator"] + fn create_select_iterator( + self: Pin<&mut SelectWrapper>, + sql: &str, + ) -> Result>; + fn database_name(self: Pin<&mut SelectWrapper>) -> Result; + + fn select_create( + sql: &str, + handle: Pin<&mut DataHandleWrapper>, + ) -> Result>; + + // ==================== WriteIteratorWrapper ==================== + + type WriteIteratorWrapper; + + fn set_column( + self: Pin<&mut WriteIteratorWrapper>, + index: usize, + name: &str, + col_type: ColumnType, + ) -> Result<()>; + fn set_number_of_columns(self: Pin<&mut WriteIteratorWrapper>, n: usize) -> Result<()>; + fn set_data(self: Pin<&mut WriteIteratorWrapper>, index: usize, value: f64) -> Result<()>; + fn set_data_string( + self: Pin<&mut WriteIteratorWrapper>, + index: usize, + value: &str, + ) -> Result<()>; + fn set_data_integer( + self: Pin<&mut WriteIteratorWrapper>, + index: usize, + value: i64, + ) -> Result<()>; + fn set_missing_value( + self: Pin<&mut WriteIteratorWrapper>, + index: usize, + value: f64, + ) -> Result<()>; + fn write_row(self: Pin<&mut WriteIteratorWrapper>) -> Result<()>; + fn close(self: Pin<&mut WriteIteratorWrapper>) -> Result<()>; + + // ==================== WriterWrapper ==================== + + type WriterWrapper; + + fn pass1(self: Pin<&mut WriterWrapper>, select: Pin<&mut SelectWrapper>) -> Result<()>; + #[cxx_name = "begin"] + fn create_write_iterator( + self: Pin<&mut WriterWrapper>, + ) -> Result>; + fn rows_buffer_size(self: &WriterWrapper) -> usize; + fn set_rows_buffer_size(self: Pin<&mut WriterWrapper>, n: usize); + fn path(self: &WriterWrapper) -> Result; + + fn writer_create(handle: Pin<&mut DataHandleWrapper>) -> Result>; + } +} + +pub use cxx::{Exception, UniquePtr}; +pub use ffi::*; + +// SAFETY: All odc wrapper types own their data with no thread-local or global mutable state. +#[allow(clippy::non_send_fields_in_send_ty)] +mod send_impls { + use super::ffi::{SelectIteratorWrapper, SelectWrapper, WriteIteratorWrapper, WriterWrapper}; + unsafe impl Send for SelectIteratorWrapper {} + unsafe impl Send for SelectWrapper {} + unsafe impl Send for WriteIteratorWrapper {} + unsafe impl Send for WriterWrapper {} +} From c6285b44fd620055c4b466f13ad72cc1528c0733 Mon Sep 17 00:00:00 2001 From: Vlad Pankratov Date: Tue, 12 May 2026 22:05:35 +0200 Subject: [PATCH 2/7] Add exception generation for ODC errors and update includes accordingly --- rust/crates/odc-sys/build.rs | 40 ++++++++++++++++++++++++---- rust/crates/odc-sys/cpp/odc_bridge.h | 2 +- rust/crates/odc-sys/src/lib.rs | 3 +++ 3 files changed, 39 insertions(+), 6 deletions(-) diff --git a/rust/crates/odc-sys/build.rs b/rust/crates/odc-sys/build.rs index 48fddcd2..50643f0e 100644 --- a/rust/crates/odc-sys/build.rs +++ b/rust/crates/odc-sys/build.rs @@ -1,5 +1,5 @@ use std::env; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; const ODC_VERSION: &str = "1.6.3"; @@ -24,17 +24,46 @@ fn main() { } } +/// Generate `odc_exceptions.{h,rs}` covering odc's own subclasses +/// (`ODBDecodeError` + its subclasses, via recursive walk in `odc/core/Exceptions.h`) +/// plus eckit's exceptions inherited from eckit-sys. +fn generate_exceptions(include: &Path) { + let out_dir = PathBuf::from(env::var("OUT_DIR").expect("OUT_DIR not set")); + + let own = vec![bindman_build::ExceptionSource { + header: include.join("odc/core/Exceptions.h"), + include_path: "odc/core/Exceptions.h".to_string(), + cpp_namespace: "odc::core".to_string(), + message_prefix: "odc".to_string(), + base_class: "eckit::Exception".to_string(), + recursive: true, + }]; + + let inherited = bindman_build::collect_dep_exception_sources(); + + bindman_build::generate_exception_bridge(&bindman_build::ExceptionBridgeConfig { + primary_namespace: "odc", + out_dir: &out_dir, + own: &own, + inherited: &inherited, + }); + + bindman_build::publish_exception_sources(&own, &out_dir); +} + #[cfg(feature = "system")] fn build_system() { let crate_dir = PathBuf::from(env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set")); + let out_dir = PathBuf::from(env::var("OUT_DIR").expect("OUT_DIR not set")); let eckit_include = env::var("DEP_ECKIT_SYS_INCLUDE").expect("DEP_ECKIT_SYS_INCLUDE not set"); - let eckit_out_dir = env::var("DEP_ECKIT_SYS_OUT_DIR").expect("DEP_ECKIT_SYS_OUT_DIR not set"); let eckit_cpp_dir = env::var("DEP_ECKIT_SYS_CPP_DIR").expect("DEP_ECKIT_SYS_CPP_DIR not set"); let (root, odc_include, lib_dir) = bindman_utils::cmake_find_package("odc", ODC_VERSION); + generate_exceptions(&odc_include); + println!("cargo:rustc-link-search=native={}", lib_dir.display()); println!("cargo:rustc-link-lib=dylib=odccore"); @@ -42,9 +71,9 @@ fn build_system() { .file(crate_dir.join("cpp/odc_bridge.cpp")) .include(&odc_include) .include(&eckit_include) - .include(&eckit_out_dir) .include(&eckit_cpp_dir) .include(crate_dir.join("cpp")) + .include(&out_dir) // for odc_exceptions.h (generated) .flag_if_supported("-std=c++17") .compile("odc_sys_bridge"); @@ -79,7 +108,6 @@ fn build_vendored() { fs::create_dir_all(&build_dir).expect("Failed to create build directory"); let eckit_root = env::var("DEP_ECKIT_SYS_ROOT").expect("DEP_ECKIT_SYS_ROOT not set"); - let eckit_out_dir = env::var("DEP_ECKIT_SYS_OUT_DIR").expect("DEP_ECKIT_SYS_OUT_DIR not set"); let eckit_cpp_dir = env::var("DEP_ECKIT_SYS_CPP_DIR").expect("DEP_ECKIT_SYS_CPP_DIR not set"); let ecbuild_src = bindman_utils::git_clone(ECBUILD_REPO, ECBUILD_TAG, &src_dir.join("ecbuild")); @@ -130,13 +158,15 @@ fn build_vendored() { PathBuf::from(env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set")); let lib_dir = bindman_utils::resolve_lib_dir(&install_dir); + generate_exceptions(&include_dir); + cxx_build::bridge("src/lib.rs") .file(crate_dir.join("cpp/odc_bridge.cpp")) .include(&include_dir) .include(format!("{eckit_root}/include")) - .include(&eckit_out_dir) .include(&eckit_cpp_dir) .include(crate_dir.join("cpp")) + .include(&out_dir) // for odc_exceptions.h (generated) .flag_if_supported("-std=c++17") .compile("odc_sys_bridge"); diff --git a/rust/crates/odc-sys/cpp/odc_bridge.h b/rust/crates/odc-sys/cpp/odc_bridge.h index 0b066561..56bf45a7 100644 --- a/rust/crates/odc-sys/cpp/odc_bridge.h +++ b/rust/crates/odc-sys/cpp/odc_bridge.h @@ -2,7 +2,7 @@ #pragma once #include "eckit_bridge.h" -#include "eckit_exceptions.h" +#include "odc_exceptions.h" #include "odc/Select.h" #include "odc/Writer.h" diff --git a/rust/crates/odc-sys/src/lib.rs b/rust/crates/odc-sys/src/lib.rs index 5f301c65..54bb116a 100644 --- a/rust/crates/odc-sys/src/lib.rs +++ b/rust/crates/odc-sys/src/lib.rs @@ -2,6 +2,9 @@ use bindman::track_cpp_api; +// Auto-generated odc Error enum + From impl +include!(concat!(env!("OUT_DIR"), "/odc_exceptions.rs")); + #[track_cpp_api( ("odc/Select.h", class = "Select"), ("odc/Writer.h", class = "Writer"), From 129fbfb341a745ce5ae14b8d637a78d1f5e90c7c Mon Sep 17 00:00:00 2001 From: Vlad Pankratov Date: Thu, 16 Jul 2026 18:30:49 +0200 Subject: [PATCH 3/7] Refactor C++ bridge for Rust FFI: update file names and remove unused code --- rust/Cargo.toml | 6 +- rust/crates/odc-sys/build.rs | 8 +- rust/crates/odc-sys/cpp/OdcBridge.cc | 148 ++++++++++++++++ rust/crates/odc-sys/cpp/OdcBridge.h | 139 +++++++++++++++ rust/crates/odc-sys/cpp/odc_bridge.cpp | 144 ---------------- rust/crates/odc-sys/cpp/odc_bridge.h | 122 ------------- rust/crates/odc-sys/src/lib.rs | 226 +++++++++++++++++-------- 7 files changed, 449 insertions(+), 344 deletions(-) create mode 100644 rust/crates/odc-sys/cpp/OdcBridge.cc create mode 100644 rust/crates/odc-sys/cpp/OdcBridge.h delete mode 100644 rust/crates/odc-sys/cpp/odc_bridge.cpp delete mode 100644 rust/crates/odc-sys/cpp/odc_bridge.h diff --git a/rust/Cargo.toml b/rust/Cargo.toml index e1f62f4c..c7d41804 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -19,9 +19,9 @@ odc-sys = { path = "crates/odc-sys" } eckit-sys = { git = "ssh://git@github.com/ecmwf/eckit.git", branch = "rust-bindings", default-features = false } # Build tools -bindman = { git = "ssh://git@github.com/ecmwf/bindman.git", branch = "generate_exception_bridge" } -bindman-build = { git = "ssh://git@github.com/ecmwf/bindman.git", branch = "generate_exception_bridge" } -bindman-utils = { git = "ssh://git@github.com/ecmwf/bindman.git", branch = "generate_exception_bridge" } +bindman = { git = "ssh://git@github.com/ecmwf/bindman.git" } +bindman-build = { git = "ssh://git@github.com/ecmwf/bindman.git" } +bindman-utils = { git = "ssh://git@github.com/ecmwf/bindman.git" } # External cxx = "1.0" diff --git a/rust/crates/odc-sys/build.rs b/rust/crates/odc-sys/build.rs index 50643f0e..9a32e924 100644 --- a/rust/crates/odc-sys/build.rs +++ b/rust/crates/odc-sys/build.rs @@ -6,8 +6,8 @@ const ODC_VERSION: &str = "1.6.3"; fn main() { println!("cargo:rerun-if-changed=build.rs"); println!("cargo:rerun-if-changed=src/lib.rs"); - println!("cargo:rerun-if-changed=cpp/odc_bridge.h"); - println!("cargo:rerun-if-changed=cpp/odc_bridge.cpp"); + println!("cargo:rerun-if-changed=cpp/OdcBridge.h"); + println!("cargo:rerun-if-changed=cpp/OdcBridge.cc"); println!("cargo:rerun-if-env-changed=ODC_DIR"); println!("cargo:rerun-if-env-changed=DOCS_RS"); @@ -68,7 +68,7 @@ fn build_system() { println!("cargo:rustc-link-lib=dylib=odccore"); cxx_build::bridge("src/lib.rs") - .file(crate_dir.join("cpp/odc_bridge.cpp")) + .file(crate_dir.join("cpp/OdcBridge.cc")) .include(&odc_include) .include(&eckit_include) .include(&eckit_cpp_dir) @@ -161,7 +161,7 @@ fn build_vendored() { generate_exceptions(&include_dir); cxx_build::bridge("src/lib.rs") - .file(crate_dir.join("cpp/odc_bridge.cpp")) + .file(crate_dir.join("cpp/OdcBridge.cc")) .include(&include_dir) .include(format!("{eckit_root}/include")) .include(&eckit_cpp_dir) diff --git a/rust/crates/odc-sys/cpp/OdcBridge.cc b/rust/crates/odc-sys/cpp/OdcBridge.cc new file mode 100644 index 00000000..14212a59 --- /dev/null +++ b/rust/crates/odc-sys/cpp/OdcBridge.cc @@ -0,0 +1,148 @@ +// odc C++ bridge implementation +#include "OdcBridge.h" + +#include "odc-sys/src/lib.rs.h" + +#include "eckit/exception/Exceptions.h" + +#include + +namespace odc_bridge { + +// ==================== FrameWrapper ==================== + +FrameWrapper::FrameWrapper(odc::api::Frame&& frame) : frame_(std::move(frame)) {} + +size_t FrameWrapper::row_count() const { + return frame_.rowCount(); +} + +size_t FrameWrapper::column_count() const { + return frame_.columnCount(); +} + +bool FrameWrapper::has_column(rust::Str name) const { + return frame_.hasColumn(std::string(name)); +} + +rust::Vec FrameWrapper::column_info() const { + rust::Vec result; + result.reserve(frame_.columnCount()); + for (const auto& info : frame_.columnInfo()) { + rust::Vec bitfield; + bitfield.reserve(info.bitfield.size()); + for (const auto& bit : info.bitfield) { + bitfield.push_back(BridgeBit{rust::String(bit.name), bit.size, bit.offset}); + } + result.push_back( + BridgeColumnInfo{rust::String(info.name), info.type, info.decodedSize, std::move(bitfield)}); + } + return result; +} + +rust::Vec FrameWrapper::properties() const { + rust::Vec result; + for (const auto& [key, value] : frame_.properties()) { + result.push_back(BridgeProperty{rust::String(key), rust::String(value)}); + } + return result; +} + +// ==================== ReaderWrapper ==================== + +ReaderWrapper::ReaderWrapper(std::unique_ptr reader) : reader_(std::move(reader)) {} + +std::unique_ptr ReaderWrapper::from_path(rust::Str path, bool aggregated, int64_t rowlimit) { + return std::make_unique( + std::make_unique(std::string(path), aggregated, static_cast(rowlimit))); +} + +std::unique_ptr ReaderWrapper::from_handle(eckit_bridge::DataHandleWrapper& handle, bool aggregated, + int64_t rowlimit) { + return std::make_unique( + std::make_unique(handle.inner(), aggregated, static_cast(rowlimit))); +} + +std::unique_ptr ReaderWrapper::next_frame() { + odc::api::Frame frame = reader_->next(); + if (!frame) { + return nullptr; + } + return std::make_unique(std::move(frame)); +} + +// ==================== DecoderWrapper ==================== + +std::unique_ptr DecoderWrapper::create() { + return std::make_unique(); +} + +void DecoderWrapper::add_column(rust::Str name, uint8_t* data, size_t nrows, size_t elem_size, size_t stride) { + names_.emplace_back(std::string(name)); + facades_.emplace_back(reinterpret_cast(data), nrows, elem_size, stride); +} + +size_t DecoderWrapper::decode(const FrameWrapper& frame, size_t nthreads) { + odc::api::Decoder decoder(names_, facades_); + decoder.decode(frame.frame_, nthreads == 0 ? 1 : nthreads); + return frame.frame_.rowCount(); +} + +// ==================== EncoderWrapper ==================== + +std::unique_ptr EncoderWrapper::create() { + return std::make_unique(); +} + +void EncoderWrapper::add_column(rust::Str name, ColumnType column_type, size_t elem_size, const uint8_t* data, + size_t nrows, size_t stride) { + columns_.push_back(odc::api::ColumnInfo{std::string(name), column_type, elem_size, {}}); + data_.emplace_back(reinterpret_cast(data), nrows, elem_size, stride); +} + +void EncoderWrapper::add_bitfield(rust::Str name, int32_t size, int32_t offset) { + if (columns_.empty()) { + throw eckit::UserError("add_bitfield called before add_column"); + } + columns_.back().bitfield.push_back(odc::api::ColumnInfo::Bit{std::string(name), size, offset}); +} + +void EncoderWrapper::set_property(rust::Str key, rust::Str value) { + properties_[std::string(key)] = std::string(value); +} + +void EncoderWrapper::encode(eckit_bridge::DataHandleWrapper& out, size_t max_rows_per_frame) { + odc::api::encode(out.inner(), columns_, data_, properties_, max_rows_per_frame); +} + +// ==================== SettingsWrapper ==================== + +void SettingsWrapper::treat_integers_as_doubles(bool flag) { + odc::api::Settings::treatIntegersAsDoubles(flag); +} + +int64_t SettingsWrapper::integer_missing_value() { + return static_cast(odc::api::Settings::integerMissingValue()); +} + +void SettingsWrapper::set_integer_missing_value(int64_t value) { + odc::api::Settings::setIntegerMissingValue(static_cast(value)); +} + +double SettingsWrapper::double_missing_value() { + return odc::api::Settings::doubleMissingValue(); +} + +void SettingsWrapper::set_double_missing_value(double value) { + odc::api::Settings::setDoubleMissingValue(value); +} + +rust::String SettingsWrapper::version() { + return rust::String(odc::api::Settings::version()); +} + +rust::String SettingsWrapper::gitsha1() { + return rust::String(odc::api::Settings::gitsha1()); +} + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/OdcBridge.h b/rust/crates/odc-sys/cpp/OdcBridge.h new file mode 100644 index 00000000..5936e5bc --- /dev/null +++ b/rust/crates/odc-sys/cpp/OdcBridge.h @@ -0,0 +1,139 @@ +// odc C++ bridge for Rust FFI +#pragma once + +#include "EckitBridge.h" +#include "odc_exceptions.h" + +#include "odc/api/ColumnType.h" +#include "odc/api/Odb.h" + +#include "rust/cxx.h" + +#include +#include +#include +#include +#include + +namespace odc_bridge { + +// ColumnType is odc::api::ColumnType — cxx static_asserts values match. +using odc::api::ColumnType; + +// cxx-generated shared structs — defined in "odc-sys/src/lib.rs.h", which +// OdcBridge.cc includes. Forward-declared here to avoid a circular include +// (the generated header includes OdcBridge.h first). +struct BridgeBit; +struct BridgeColumnInfo; +struct BridgeProperty; + +// ==================== FrameWrapper ==================== + +/// Owns an `odc::api::Frame` — a viewport onto a chunk of contiguous, +/// compatible data (possibly a logical frame aggregating several physical +/// frames). The frame reads lazily from the Reader's stream, so it must not +/// outlive its ReaderWrapper — enforced on the Rust side. +class FrameWrapper { + odc::api::Frame frame_; + + friend class DecoderWrapper; + +public: + + explicit FrameWrapper(odc::api::Frame&& frame); + + size_t row_count() const; + size_t column_count() const; + bool has_column(rust::Str name) const; + rust::Vec column_info() const; + rust::Vec properties() const; +}; + +// ==================== ReaderWrapper ==================== + +/// Wraps `odc::api::Reader` — owns the ODB-2 stream, yields frames. +class ReaderWrapper { + std::unique_ptr reader_; + +public: + + explicit ReaderWrapper(std::unique_ptr reader); + + static std::unique_ptr from_path(rust::Str path, bool aggregated, int64_t rowlimit); + + /// Does not take ownership: the DataHandleWrapper must outlive this + /// reader (and every frame it yields) — enforced on the Rust side. + /// The handle must not be open; the reader opens it for reading. + static std::unique_ptr from_handle(eckit_bridge::DataHandleWrapper& handle, bool aggregated, + int64_t rowlimit); + + /// Returns nullptr when the stream is exhausted. + std::unique_ptr next_frame(); +}; + +// ==================== DecoderWrapper ==================== + +/// Accumulates per-column decode targets (StridedData facades over +/// Rust-owned buffers), then runs `odc::api::Decoder::decode`. +class DecoderWrapper { + std::vector names_; + std::vector facades_; + +public: + + static std::unique_ptr create(); + + /// `data` must point to caller-owned, 8-byte-aligned memory of at least + /// nrows * stride bytes, valid until decode() returns. + void add_column(rust::Str name, uint8_t* data, size_t nrows, size_t elem_size, size_t stride); + + /// Decode the frame into the registered buffers; returns rows decoded. + size_t decode(const FrameWrapper& frame, size_t nthreads); +}; + +// ==================== EncoderWrapper ==================== + +/// Accumulates column specs + ConstStridedData facades over Rust-owned +/// buffers, then calls the `odc::api::encode` free function. +class EncoderWrapper { + std::vector columns_; + std::vector data_; + std::map properties_; + +public: + + static std::unique_ptr create(); + + /// `data` must point to caller-owned memory of at least nrows * stride + /// bytes, valid until encode() returns. + void add_column(rust::Str name, ColumnType column_type, size_t elem_size, const uint8_t* data, size_t nrows, + size_t stride); + + /// Append a bit group to the most recently added column. + void add_bitfield(rust::Str name, int32_t size, int32_t offset); + + void set_property(rust::Str key, rust::Str value); + + /// Encode all columns to an (already open) data handle. + void encode(eckit_bridge::DataHandleWrapper& out, size_t max_rows_per_frame); +}; + +// ==================== SettingsWrapper ==================== + +/// Wraps `odc::api::Settings` — process-global settings and version info. +/// Static-only; never instantiated. +class SettingsWrapper { +public: + + SettingsWrapper() = delete; + + static void treat_integers_as_doubles(bool flag); + static int64_t integer_missing_value(); + static void set_integer_missing_value(int64_t value); + static double double_missing_value(); + static void set_double_missing_value(double value); + static rust::String version(); + static rust::String gitsha1(); +}; + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/odc_bridge.cpp b/rust/crates/odc-sys/cpp/odc_bridge.cpp deleted file mode 100644 index 80e22943..00000000 --- a/rust/crates/odc-sys/cpp/odc_bridge.cpp +++ /dev/null @@ -1,144 +0,0 @@ -// odc C++ bridge implementation -#include "odc_bridge.h" - -#include "eckit/io/MemoryHandle.h" - -#include - -namespace odc_bridge { - -// ==================== SelectIteratorWrapper ==================== - -SelectIteratorWrapper::SelectIteratorWrapper(odc::Select& select) : current_(select.begin()), end_(select.end()) {} - -SelectIteratorWrapper::SelectIteratorWrapper(odc::Select::iterator current, odc::Select::iterator end) : - current_(std::move(current)), end_(std::move(end)) {} - -bool SelectIteratorWrapper::valid() { - return current_ != end_; -} - -void SelectIteratorWrapper::advance() { - ++current_; -} - -size_t SelectIteratorWrapper::column_count() const { - return current_->columns().size(); -} - -rust::String SelectIteratorWrapper::column_name(size_t idx) const { - return rust::String(current_->columns()[idx]->name()); -} - -ColumnType SelectIteratorWrapper::column_type(size_t idx) const { - return current_->columns()[idx]->type(); -} - -double SelectIteratorWrapper::data(size_t idx) const { - return current_->data(idx); -} - -rust::String SelectIteratorWrapper::data_string(size_t idx) { - return rust::String(current_->string(idx)); -} - -int64_t SelectIteratorWrapper::data_integer(size_t idx) { - return current_->integer(idx); -} - -// ==================== SelectWrapper ==================== - -SelectWrapper::SelectWrapper(rust::Str sql, eckit_bridge::DataHandleWrapper& handle) : - select_(std::make_unique(std::string(sql), handle.inner())) {} - -std::unique_ptr SelectWrapper::begin() { - return std::make_unique(*select_); -} - -std::unique_ptr SelectWrapper::createSelectIterator(rust::Str sql) { - auto* it = select_->createSelectIterator(std::string(sql)); - it->next(); - return std::make_unique(odc::Select::iterator(it), select_->end()); -} - -rust::String SelectWrapper::database_name() { - return rust::String(select_->database().name()); -} - -std::unique_ptr select_create(rust::Str sql, eckit_bridge::DataHandleWrapper& handle) { - return std::make_unique(sql, handle); -} - -// ==================== WriteIteratorWrapper ==================== - -WriteIteratorWrapper::WriteIteratorWrapper(odc::Writer<>::iterator iter) : iter_(std::move(iter)) {} - -void WriteIteratorWrapper::set_column(size_t index, rust::Str name, ColumnType col_type) { - iter_->setColumn(index, std::string(name), col_type); -} - -void WriteIteratorWrapper::set_number_of_columns(size_t n) { - iter_->setNumberOfColumns(n); -} - -void WriteIteratorWrapper::set_data(size_t index, double value) { - iter_->data(index) = value; -} - -void WriteIteratorWrapper::set_data_string(size_t index, rust::Str value) { - size_t maxlen = sizeof(double) * iter_->columns()[index]->dataSizeDoubles(); - ::strncpy(reinterpret_cast(&iter_->data(index)), std::string(value).c_str(), maxlen); -} - -void WriteIteratorWrapper::set_data_integer(size_t index, int64_t value) { - iter_->data(index) = static_cast(value); -} - -void WriteIteratorWrapper::set_missing_value(size_t index, double value) { - iter_->missingValue(index, value); -} - -void WriteIteratorWrapper::write_row() { - ++iter_; -} - -void WriteIteratorWrapper::close() { - iter_->close(); -} - -// ==================== WriterWrapper ==================== - -WriterWrapper::WriterWrapper(eckit_bridge::DataHandleWrapper& handle) : - writer_(std::make_unique>(handle.inner())), outit_(writer_->begin()) {} - -std::unique_ptr WriterWrapper::begin() { - return std::make_unique(writer_->begin()); -} - -void WriterWrapper::pass1(SelectWrapper& select) { - auto it = select.select_->begin(); - auto end = select.select_->end(); - outit_->pass1(it, end); -} - -size_t WriterWrapper::rows_buffer_size() const { - return writer_->rowsBufferSize(); -} - -void WriterWrapper::set_rows_buffer_size(size_t n) { - writer_->rowsBufferSize(n); -} - -// Note: data_handle() not exposed — Writer owns the DataHandle internally -// and it's the same one passed to the constructor. Access it via the -// original DataHandleWrapper on the Rust side. - -rust::String WriterWrapper::path() const { - return rust::String(writer_->path()); -} - -std::unique_ptr writer_create(eckit_bridge::DataHandleWrapper& handle) { - return std::make_unique(handle); -} - -} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/odc_bridge.h b/rust/crates/odc-sys/cpp/odc_bridge.h deleted file mode 100644 index 56bf45a7..00000000 --- a/rust/crates/odc-sys/cpp/odc_bridge.h +++ /dev/null @@ -1,122 +0,0 @@ -// odc C++ bridge for Rust FFI -#pragma once - -#include "eckit_bridge.h" -#include "odc_exceptions.h" - -#include "odc/Select.h" -#include "odc/Writer.h" -#include "odc/api/ColumnType.h" - -#include "rust/cxx.h" - -#include -#include -#include - -namespace odc_bridge { - -// ColumnType is odc::api::ColumnType — cxx static_asserts values match. -using odc::api::ColumnType; - -// ==================== SelectIteratorWrapper ==================== - -/// Wraps `odc::Select::iterator` pair for row-by-row iteration. -class SelectIteratorWrapper { - odc::Select::iterator current_; - odc::Select::iterator end_; - -public: - - SelectIteratorWrapper(odc::Select& select); - SelectIteratorWrapper(odc::Select::iterator current, odc::Select::iterator end); - - bool valid(); - void advance(); - - size_t column_count() const; - rust::String column_name(size_t idx) const; - ColumnType column_type(size_t idx) const; - double data(size_t idx) const; - rust::String data_string(size_t idx); - int64_t data_integer(size_t idx); -}; - -// ==================== SelectWrapper ==================== - -/// Wraps `odc::Select` — owns the query, produces iterators. -class SelectWrapper { - std::unique_ptr select_; - - friend class WriterWrapper; - -public: - - SelectWrapper(rust::Str sql, eckit_bridge::DataHandleWrapper& handle); - - std::unique_ptr begin(); - std::unique_ptr createSelectIterator(rust::Str sql); - rust::String database_name(); -}; - -std::unique_ptr select_create(rust::Str sql, eckit_bridge::DataHandleWrapper& handle); - -// ==================== WriteIteratorWrapper ==================== - -/// Wraps `odc::Writer<>::iterator` for row-by-row writing. -class WriteIteratorWrapper { - odc::Writer<>::iterator iter_; - -public: - - explicit WriteIteratorWrapper(odc::Writer<>::iterator iter); - - /// Define a column. Must be called before writing any rows. - void set_column(size_t index, rust::Str name, ColumnType col_type); - - /// Set number of columns. - void set_number_of_columns(size_t n); - - /// Set a double value at column index for the current row. - void set_data(size_t index, double value); - - /// Set a string value at column index for the current row. - void set_data_string(size_t index, rust::Str value); - - /// Set an integer value at column index for the current row. - void set_data_integer(size_t index, int64_t value); - - /// Set the missing value for a column. - void set_missing_value(size_t index, double value); - - /// Write the current row (advances the iterator). - void write_row(); - - /// Close the writer. - void close(); -}; - -// ==================== WriterWrapper ==================== - -/// Wraps `odc::Writer<>` — writes filtered ODB data via pass1. -class WriterWrapper { - std::unique_ptr> writer_; - odc::Writer<>::iterator outit_; - -public: - - explicit WriterWrapper(eckit_bridge::DataHandleWrapper& handle); - - void pass1(SelectWrapper& select); - - /// Get a write iterator for row-by-row writing. - std::unique_ptr begin(); - - size_t rows_buffer_size() const; - void set_rows_buffer_size(size_t n); - rust::String path() const; -}; - -std::unique_ptr writer_create(eckit_bridge::DataHandleWrapper& handle); - -} // namespace odc_bridge diff --git a/rust/crates/odc-sys/src/lib.rs b/rust/crates/odc-sys/src/lib.rs index 54bb116a..b837b45f 100644 --- a/rust/crates/odc-sys/src/lib.rs +++ b/rust/crates/odc-sys/src/lib.rs @@ -1,4 +1,7 @@ //! FFI bindings to ECMWF odc (ODB-2 encoder/decoder) library. +//! +//! Wraps the public C++ API (`odc::api`): `Reader` → `Frame` → +//! `Decoder`/`encode()`, plus the global `Settings`. use bindman::track_cpp_api; @@ -6,9 +9,11 @@ use bindman::track_cpp_api; include!(concat!(env!("OUT_DIR"), "/odc_exceptions.rs")); #[track_cpp_api( - ("odc/Select.h", class = "Select"), - ("odc/Writer.h", class = "Writer"), - ignore = ["end", "dataHandle"] + ("odc/api/Odb.h", class = "Reader"), + ("odc/api/Odb.h", class = "Frame"), + ("odc/api/Odb.h", class = "Decoder"), + ("odc/api/Odb.h", class = "Settings"), + ignore = ["offset", "length", "filter", "encodedData", "span", "slice"] )] #[cxx::bridge(namespace = "odc_bridge")] pub mod ffi { @@ -30,8 +35,36 @@ pub mod ffi { Double = 5, } + /// A bit group within a bitfield column. + #[derive(Debug, Clone, PartialEq, Eq)] + struct BridgeBit { + name: String, + /// Bit group size in bits. + size: i32, + /// Bit group offset in bits. + offset: i32, + } + + /// Metadata for one column of a frame. + #[derive(Debug, Clone)] + struct BridgeColumnInfo { + name: String, + column_type: ColumnType, + /// Size of a single decoded value in bytes (always a multiple of 8). + decoded_size: usize, + /// Bit groups — non-empty only for bitfield columns. + bitfield: Vec, + } + + /// A key/value property encoded in a frame. + #[derive(Debug, Clone, PartialEq, Eq)] + struct BridgeProperty { + key: String, + value: String, + } + unsafe extern "C++" { - include!("odc_bridge.h"); + include!("OdcBridge.h"); // Verify ColumnType matches C++ odc::api::ColumnType at compile time #[namespace = "odc::api"] @@ -41,93 +74,144 @@ pub mod ffi { #[namespace = "eckit_bridge"] type DataHandleWrapper = eckit_sys::DataHandleWrapper; - // ==================== SelectIteratorWrapper ==================== + // ==================== ReaderWrapper ==================== + + type ReaderWrapper; - type SelectIteratorWrapper; + /// Open an ODB-2 file for reading. + #[Self = "ReaderWrapper"] + fn from_path( + path: &str, + aggregated: bool, + rowlimit: i64, + ) -> Result>; - #[must_use] - fn valid(self: Pin<&mut SelectIteratorWrapper>) -> bool; - fn advance(self: Pin<&mut SelectIteratorWrapper>) -> Result<()>; - fn column_count(self: &SelectIteratorWrapper) -> usize; - fn column_name(self: &SelectIteratorWrapper, idx: usize) -> Result; - fn column_type(self: &SelectIteratorWrapper, idx: usize) -> Result; - fn data(self: &SelectIteratorWrapper, idx: usize) -> Result; - fn data_string(self: Pin<&mut SelectIteratorWrapper>, idx: usize) -> Result; - fn data_integer(self: Pin<&mut SelectIteratorWrapper>, idx: usize) -> Result; + /// Read from an eckit data handle. Does NOT take ownership: the + /// handle must be unopened and must outlive the reader and every + /// frame it yields. + #[Self = "ReaderWrapper"] + fn from_handle( + handle: Pin<&mut DataHandleWrapper>, + aggregated: bool, + rowlimit: i64, + ) -> Result>; - // ==================== SelectWrapper ==================== + /// Next frame in the stream; null when exhausted. + fn next_frame(self: Pin<&mut ReaderWrapper>) -> Result>; - type SelectWrapper; + // ==================== FrameWrapper ==================== - fn begin(self: Pin<&mut SelectWrapper>) -> Result>; - #[cxx_name = "createSelectIterator"] - fn create_select_iterator( - self: Pin<&mut SelectWrapper>, - sql: &str, - ) -> Result>; - fn database_name(self: Pin<&mut SelectWrapper>) -> Result; + type FrameWrapper; - fn select_create( - sql: &str, - handle: Pin<&mut DataHandleWrapper>, - ) -> Result>; + fn row_count(self: &FrameWrapper) -> usize; + fn column_count(self: &FrameWrapper) -> usize; + fn has_column(self: &FrameWrapper, name: &str) -> bool; + fn column_info(self: &FrameWrapper) -> Result>; + fn properties(self: &FrameWrapper) -> Result>; - // ==================== WriteIteratorWrapper ==================== + // ==================== DecoderWrapper ==================== - type WriteIteratorWrapper; + type DecoderWrapper; - fn set_column( - self: Pin<&mut WriteIteratorWrapper>, - index: usize, + #[Self = "DecoderWrapper"] + fn create() -> UniquePtr; + + /// Register a decode target for the named column. + /// + /// # Safety + /// + /// `data` must be 8-byte-aligned, valid for `nrows * stride` bytes, + /// and must not be dropped or aliased until `decode` returns. + unsafe fn add_column( + self: Pin<&mut DecoderWrapper>, name: &str, - col_type: ColumnType, - ) -> Result<()>; - fn set_number_of_columns(self: Pin<&mut WriteIteratorWrapper>, n: usize) -> Result<()>; - fn set_data(self: Pin<&mut WriteIteratorWrapper>, index: usize, value: f64) -> Result<()>; - fn set_data_string( - self: Pin<&mut WriteIteratorWrapper>, - index: usize, - value: &str, - ) -> Result<()>; - fn set_data_integer( - self: Pin<&mut WriteIteratorWrapper>, - index: usize, - value: i64, - ) -> Result<()>; - fn set_missing_value( - self: Pin<&mut WriteIteratorWrapper>, - index: usize, - value: f64, + data: *mut u8, + nrows: usize, + elem_size: usize, + stride: usize, + ); + + /// Decode the frame into the registered buffers; returns rows decoded. + fn decode( + self: Pin<&mut DecoderWrapper>, + frame: &FrameWrapper, + nthreads: usize, + ) -> Result; + + // ==================== EncoderWrapper ==================== + + type EncoderWrapper; + + #[Self = "EncoderWrapper"] + fn create() -> UniquePtr; + + /// Register a source column for encoding. + /// + /// # Safety + /// + /// `data` must be valid for `nrows * stride` bytes and must not be + /// dropped until `encode` returns. + unsafe fn add_column( + self: Pin<&mut EncoderWrapper>, + name: &str, + column_type: ColumnType, + elem_size: usize, + data: *const u8, + nrows: usize, + stride: usize, + ); + + /// Append a bit group to the most recently added column. + fn add_bitfield( + self: Pin<&mut EncoderWrapper>, + name: &str, + size: i32, + offset: i32, ) -> Result<()>; - fn write_row(self: Pin<&mut WriteIteratorWrapper>) -> Result<()>; - fn close(self: Pin<&mut WriteIteratorWrapper>) -> Result<()>; - // ==================== WriterWrapper ==================== + fn set_property(self: Pin<&mut EncoderWrapper>, key: &str, value: &str); - type WriterWrapper; - - fn pass1(self: Pin<&mut WriterWrapper>, select: Pin<&mut SelectWrapper>) -> Result<()>; - #[cxx_name = "begin"] - fn create_write_iterator( - self: Pin<&mut WriterWrapper>, - ) -> Result>; - fn rows_buffer_size(self: &WriterWrapper) -> usize; - fn set_rows_buffer_size(self: Pin<&mut WriterWrapper>, n: usize); - fn path(self: &WriterWrapper) -> Result; + /// Encode all registered columns to an (already open) data handle. + fn encode( + self: Pin<&mut EncoderWrapper>, + out: Pin<&mut DataHandleWrapper>, + max_rows_per_frame: usize, + ) -> Result<()>; - fn writer_create(handle: Pin<&mut DataHandleWrapper>) -> Result>; + // ==================== SettingsWrapper (process-global) ==================== + + type SettingsWrapper; + + /// Whether INTEGER/BITFIELD columns decode as doubles (true, odc + /// default) or as int64 (false). + #[Self = "SettingsWrapper"] + fn treat_integers_as_doubles(flag: bool); + #[Self = "SettingsWrapper"] + fn integer_missing_value() -> i64; + #[Self = "SettingsWrapper"] + fn set_integer_missing_value(value: i64); + #[Self = "SettingsWrapper"] + fn double_missing_value() -> f64; + #[Self = "SettingsWrapper"] + fn set_double_missing_value(value: f64); + #[Self = "SettingsWrapper"] + fn version() -> String; + #[Self = "SettingsWrapper"] + fn gitsha1() -> String; } } pub use cxx::{Exception, UniquePtr}; pub use ffi::*; -// SAFETY: All odc wrapper types own their data with no thread-local or global mutable state. +// SAFETY: All odc wrapper types have no thread affinity or thread-local +// state. Frames share the reader's underlying stream, but that access is +// serialized C++-side by odc::core::ThreadSharedDataHandle. #[allow(clippy::non_send_fields_in_send_ty)] mod send_impls { - use super::ffi::{SelectIteratorWrapper, SelectWrapper, WriteIteratorWrapper, WriterWrapper}; - unsafe impl Send for SelectIteratorWrapper {} - unsafe impl Send for SelectWrapper {} - unsafe impl Send for WriteIteratorWrapper {} - unsafe impl Send for WriterWrapper {} + use super::ffi::{DecoderWrapper, EncoderWrapper, FrameWrapper, ReaderWrapper}; + unsafe impl Send for ReaderWrapper {} + unsafe impl Send for FrameWrapper {} + unsafe impl Send for DecoderWrapper {} + unsafe impl Send for EncoderWrapper {} } From fd30d647010bbdeec8e647b963db2c8824bf95e7 Mon Sep 17 00:00:00 2001 From: Vlad Pankratov Date: Thu, 16 Jul 2026 18:40:22 +0200 Subject: [PATCH 4/7] Update bindman dependencies to specific revision --- rust/Cargo.toml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/rust/Cargo.toml b/rust/Cargo.toml index c7d41804..3b5881b6 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -19,9 +19,9 @@ odc-sys = { path = "crates/odc-sys" } eckit-sys = { git = "ssh://git@github.com/ecmwf/eckit.git", branch = "rust-bindings", default-features = false } # Build tools -bindman = { git = "ssh://git@github.com/ecmwf/bindman.git" } -bindman-build = { git = "ssh://git@github.com/ecmwf/bindman.git" } -bindman-utils = { git = "ssh://git@github.com/ecmwf/bindman.git" } +bindman = { git = "ssh://git@github.com/ecmwf/bindman.git", rev = "47edf68" } +bindman-build = { git = "ssh://git@github.com/ecmwf/bindman.git", rev = "47edf68" } +bindman-utils = { git = "ssh://git@github.com/ecmwf/bindman.git", rev = "47edf68" } # External cxx = "1.0" From fc6db66364d39afadd3eaa63a0ba82c0cff0d82d Mon Sep 17 00:00:00 2001 From: Vlad Pankratov Date: Thu, 16 Jul 2026 18:45:36 +0200 Subject: [PATCH 5/7] Add Rust CI workflow to main CI configuration for formatting and testing --- .github/workflows/ci-rust.yml | 79 ----------------------------------- .github/workflows/ci.yml | 12 ++++++ 2 files changed, 12 insertions(+), 79 deletions(-) delete mode 100644 .github/workflows/ci-rust.yml diff --git a/.github/workflows/ci-rust.yml b/.github/workflows/ci-rust.yml deleted file mode 100644 index 48efac09..00000000 --- a/.github/workflows/ci-rust.yml +++ /dev/null @@ -1,79 +0,0 @@ -name: rust - -on: - push: - branches: - - 'master' - - 'develop' - - 'rust-bindings' - tags-ignore: - - '**' - paths: - - 'rust/**' - - '.github/workflows/ci-rust.yml' - - pull_request: - paths: - - 'rust/**' - - '.github/workflows/ci-rust.yml' - - workflow_dispatch: ~ - -env: - CARGO_TERM_COLOR: always - CARGO_NET_GIT_FETCH_WITH_CLI: "true" - -jobs: - fmt: - name: fmt - runs-on: ubuntu-latest - defaults: - run: - working-directory: rust - steps: - - uses: actions/checkout@v4 - - - uses: dtolnay/rust-toolchain@stable - with: - components: rustfmt - - - name: Format check - run: cargo fmt --check - - clippy: - name: clippy - if: ${{ !github.event.pull_request.head.repo.fork }} - runs-on: ubuntu-latest - defaults: - run: - working-directory: rust - steps: - - uses: actions/checkout@v4 - - - name: Configure git for private repos - run: git config --global url."https://x-access-token:${{ secrets.GH_REPO_READ_TOKEN }}@github.com/".insteadOf "ssh://git@github.com/" - - - uses: dtolnay/rust-toolchain@stable - with: - components: clippy - - - name: Clippy - run: cargo clippy --features vendored --all-targets -- -D warnings - - test: - name: test - if: ${{ !github.event.pull_request.head.repo.fork }} - runs-on: ubuntu-latest - defaults: - run: - working-directory: rust - steps: - - uses: actions/checkout@v4 - - - name: Configure git for private repos - run: git config --global url."https://x-access-token:${{ secrets.GH_REPO_READ_TOKEN }}@github.com/".insteadOf "ssh://git@github.com/" - - - uses: dtolnay/rust-toolchain@stable - - - name: Test - run: cargo test --features vendored diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cc128dab..56bb5e2a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,6 +20,18 @@ on: types: [labeled] jobs: + # Run Rust CI (fmt + clippy + test) on the rust/ workspace + ci-rust: + name: ci-rust + if: ${{ !github.event.pull_request.head.repo.fork && github.event.action != 'labeled' || github.event.label.name == 'approved-for-ci' }} + uses: ecmwf/reusable-workflows/.github/workflows/ci-rust.yml@main + with: + manifest-path: rust/Cargo.toml + features: --features vendored + run-doc: false + secrets: + private_repos_token: ${{ secrets.GH_REPO_READ_TOKEN }} + # Run CI including downstream packages on self-hosted runners downstream-ci: name: downstream-ci From 9525b4f35010dd55c70192af1220fb51c115606c Mon Sep 17 00:00:00 2001 From: Vlad Pankratov Date: Thu, 16 Jul 2026 19:04:25 +0200 Subject: [PATCH 6/7] Refactor build script to include new C++ wrappers for decoding and encoding --- rust/crates/odc-sys/build.rs | 23 +++- rust/crates/odc-sys/cpp/DecoderWrapper.cc | 25 ++++ rust/crates/odc-sys/cpp/DecoderWrapper.h | 36 +++++ rust/crates/odc-sys/cpp/EncoderWrapper.cc | 37 +++++ rust/crates/odc-sys/cpp/EncoderWrapper.h | 48 +++++++ rust/crates/odc-sys/cpp/FrameWrapper.cc | 47 +++++++ rust/crates/odc-sys/cpp/FrameWrapper.h | 39 ++++++ rust/crates/odc-sys/cpp/OdcBridge.cc | 148 -------------------- rust/crates/odc-sys/cpp/OdcBridge.h | 152 ++------------------- rust/crates/odc-sys/cpp/ReaderWrapper.cc | 32 +++++ rust/crates/odc-sys/cpp/ReaderWrapper.h | 36 +++++ rust/crates/odc-sys/cpp/SettingsWrapper.cc | 38 ++++++ rust/crates/odc-sys/cpp/SettingsWrapper.h | 26 ++++ rust/crates/odc-sys/src/lib.rs | 4 + 14 files changed, 403 insertions(+), 288 deletions(-) create mode 100644 rust/crates/odc-sys/cpp/DecoderWrapper.cc create mode 100644 rust/crates/odc-sys/cpp/DecoderWrapper.h create mode 100644 rust/crates/odc-sys/cpp/EncoderWrapper.cc create mode 100644 rust/crates/odc-sys/cpp/EncoderWrapper.h create mode 100644 rust/crates/odc-sys/cpp/FrameWrapper.cc create mode 100644 rust/crates/odc-sys/cpp/FrameWrapper.h delete mode 100644 rust/crates/odc-sys/cpp/OdcBridge.cc create mode 100644 rust/crates/odc-sys/cpp/ReaderWrapper.cc create mode 100644 rust/crates/odc-sys/cpp/ReaderWrapper.h create mode 100644 rust/crates/odc-sys/cpp/SettingsWrapper.cc create mode 100644 rust/crates/odc-sys/cpp/SettingsWrapper.h diff --git a/rust/crates/odc-sys/build.rs b/rust/crates/odc-sys/build.rs index 9a32e924..a282b23a 100644 --- a/rust/crates/odc-sys/build.rs +++ b/rust/crates/odc-sys/build.rs @@ -7,7 +7,16 @@ fn main() { println!("cargo:rerun-if-changed=build.rs"); println!("cargo:rerun-if-changed=src/lib.rs"); println!("cargo:rerun-if-changed=cpp/OdcBridge.h"); - println!("cargo:rerun-if-changed=cpp/OdcBridge.cc"); + println!("cargo:rerun-if-changed=cpp/DecoderWrapper.h"); + println!("cargo:rerun-if-changed=cpp/DecoderWrapper.cc"); + println!("cargo:rerun-if-changed=cpp/EncoderWrapper.h"); + println!("cargo:rerun-if-changed=cpp/EncoderWrapper.cc"); + println!("cargo:rerun-if-changed=cpp/FrameWrapper.h"); + println!("cargo:rerun-if-changed=cpp/FrameWrapper.cc"); + println!("cargo:rerun-if-changed=cpp/ReaderWrapper.h"); + println!("cargo:rerun-if-changed=cpp/ReaderWrapper.cc"); + println!("cargo:rerun-if-changed=cpp/SettingsWrapper.h"); + println!("cargo:rerun-if-changed=cpp/SettingsWrapper.cc"); println!("cargo:rerun-if-env-changed=ODC_DIR"); println!("cargo:rerun-if-env-changed=DOCS_RS"); @@ -68,7 +77,11 @@ fn build_system() { println!("cargo:rustc-link-lib=dylib=odccore"); cxx_build::bridge("src/lib.rs") - .file(crate_dir.join("cpp/OdcBridge.cc")) + .file(crate_dir.join("cpp/DecoderWrapper.cc")) + .file(crate_dir.join("cpp/EncoderWrapper.cc")) + .file(crate_dir.join("cpp/FrameWrapper.cc")) + .file(crate_dir.join("cpp/ReaderWrapper.cc")) + .file(crate_dir.join("cpp/SettingsWrapper.cc")) .include(&odc_include) .include(&eckit_include) .include(&eckit_cpp_dir) @@ -161,7 +174,11 @@ fn build_vendored() { generate_exceptions(&include_dir); cxx_build::bridge("src/lib.rs") - .file(crate_dir.join("cpp/OdcBridge.cc")) + .file(crate_dir.join("cpp/DecoderWrapper.cc")) + .file(crate_dir.join("cpp/EncoderWrapper.cc")) + .file(crate_dir.join("cpp/FrameWrapper.cc")) + .file(crate_dir.join("cpp/ReaderWrapper.cc")) + .file(crate_dir.join("cpp/SettingsWrapper.cc")) .include(&include_dir) .include(format!("{eckit_root}/include")) .include(&eckit_cpp_dir) diff --git a/rust/crates/odc-sys/cpp/DecoderWrapper.cc b/rust/crates/odc-sys/cpp/DecoderWrapper.cc new file mode 100644 index 00000000..c3f16cf8 --- /dev/null +++ b/rust/crates/odc-sys/cpp/DecoderWrapper.cc @@ -0,0 +1,25 @@ +// odc Decoder bridge — implementation. + +#include "DecoderWrapper.h" +#include "odc-sys/src/lib.rs.h" + +#include + +namespace odc_bridge { + +std::unique_ptr DecoderWrapper::create() { + return std::make_unique(); +} + +void DecoderWrapper::add_column(rust::Str name, uint8_t* data, size_t nrows, size_t elem_size, size_t stride) { + names_.emplace_back(std::string(name)); + facades_.emplace_back(reinterpret_cast(data), nrows, elem_size, stride); +} + +size_t DecoderWrapper::decode(const FrameWrapper& frame, size_t nthreads) { + odc::api::Decoder decoder(names_, facades_); + decoder.decode(frame.frame_, nthreads == 0 ? 1 : nthreads); + return frame.frame_.rowCount(); +} + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/DecoderWrapper.h b/rust/crates/odc-sys/cpp/DecoderWrapper.h new file mode 100644 index 00000000..5f059291 --- /dev/null +++ b/rust/crates/odc-sys/cpp/DecoderWrapper.h @@ -0,0 +1,36 @@ +// odc Decoder bridge — wraps `odc::api::Decoder`. +#pragma once + +#include "FrameWrapper.h" + +#include "odc/api/Odb.h" + +#include "rust/cxx.h" + +#include +#include +#include +#include +#include + +namespace odc_bridge { + +/// Accumulates per-column decode targets (StridedData facades over +/// Rust-owned buffers), then runs `odc::api::Decoder::decode`. +class DecoderWrapper { + std::vector names_; + std::vector facades_; + +public: + + static std::unique_ptr create(); + + /// `data` must point to caller-owned, 8-byte-aligned memory of at least + /// nrows * stride bytes, valid until decode() returns. + void add_column(rust::Str name, uint8_t* data, size_t nrows, size_t elem_size, size_t stride); + + /// Decode the frame into the registered buffers; returns rows decoded. + size_t decode(const FrameWrapper& frame, size_t nthreads); +}; + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/EncoderWrapper.cc b/rust/crates/odc-sys/cpp/EncoderWrapper.cc new file mode 100644 index 00000000..b8999a63 --- /dev/null +++ b/rust/crates/odc-sys/cpp/EncoderWrapper.cc @@ -0,0 +1,37 @@ +// odc encode bridge — implementation. + +#include "EncoderWrapper.h" +#include "odc-sys/src/lib.rs.h" + +#include "eckit/exception/Exceptions.h" + +#include + +namespace odc_bridge { + +std::unique_ptr EncoderWrapper::create() { + return std::make_unique(); +} + +void EncoderWrapper::add_column(rust::Str name, ColumnType column_type, size_t elem_size, const uint8_t* data, + size_t nrows, size_t stride) { + columns_.push_back(odc::api::ColumnInfo{std::string(name), column_type, elem_size, {}}); + data_.emplace_back(reinterpret_cast(data), nrows, elem_size, stride); +} + +void EncoderWrapper::add_bitfield(rust::Str name, int32_t size, int32_t offset) { + if (columns_.empty()) { + throw eckit::UserError("add_bitfield called before add_column"); + } + columns_.back().bitfield.push_back(odc::api::ColumnInfo::Bit{std::string(name), size, offset}); +} + +void EncoderWrapper::set_property(rust::Str key, rust::Str value) { + properties_[std::string(key)] = std::string(value); +} + +void EncoderWrapper::encode(eckit_bridge::DataHandleWrapper& out, size_t max_rows_per_frame) { + odc::api::encode(out.inner(), columns_, data_, properties_, max_rows_per_frame); +} + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/EncoderWrapper.h b/rust/crates/odc-sys/cpp/EncoderWrapper.h new file mode 100644 index 00000000..a6f5035a --- /dev/null +++ b/rust/crates/odc-sys/cpp/EncoderWrapper.h @@ -0,0 +1,48 @@ +// odc encode bridge — wraps the `odc::api::encode` free function. +#pragma once + +#include "DataHandleWrapper.h" + +#include "odc/api/ColumnType.h" +#include "odc/api/Odb.h" + +#include "rust/cxx.h" + +#include +#include +#include +#include +#include +#include + +namespace odc_bridge { + +// ColumnType is odc::api::ColumnType — cxx static_asserts values match. +using odc::api::ColumnType; + +/// Accumulates column specs + ConstStridedData facades over Rust-owned +/// buffers, then calls the `odc::api::encode` free function. +class EncoderWrapper { + std::vector columns_; + std::vector data_; + std::map properties_; + +public: + + static std::unique_ptr create(); + + /// `data` must point to caller-owned memory of at least nrows * stride + /// bytes, valid until encode() returns. + void add_column(rust::Str name, ColumnType column_type, size_t elem_size, const uint8_t* data, size_t nrows, + size_t stride); + + /// Append a bit group to the most recently added column. + void add_bitfield(rust::Str name, int32_t size, int32_t offset); + + void set_property(rust::Str key, rust::Str value); + + /// Encode all columns to an (already open) data handle. + void encode(eckit_bridge::DataHandleWrapper& out, size_t max_rows_per_frame); +}; + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/FrameWrapper.cc b/rust/crates/odc-sys/cpp/FrameWrapper.cc new file mode 100644 index 00000000..4d494a4e --- /dev/null +++ b/rust/crates/odc-sys/cpp/FrameWrapper.cc @@ -0,0 +1,47 @@ +// odc Frame bridge — implementation. + +#include "FrameWrapper.h" +#include "odc-sys/src/lib.rs.h" + +#include +#include + +namespace odc_bridge { + +FrameWrapper::FrameWrapper(odc::api::Frame&& frame) : frame_(std::move(frame)) {} + +size_t FrameWrapper::row_count() const { + return frame_.rowCount(); +} + +size_t FrameWrapper::column_count() const { + return frame_.columnCount(); +} + +bool FrameWrapper::has_column(rust::Str name) const { + return frame_.hasColumn(std::string(name)); +} + +rust::Vec FrameWrapper::column_info() const { + rust::Vec result; + result.reserve(frame_.columnCount()); + for (const auto& info : frame_.columnInfo()) { + rust::Vec bitfield; + bitfield.reserve(info.bitfield.size()); + for (const auto& bit : info.bitfield) { + bitfield.push_back(BridgeBit{rust::String(bit.name), bit.size, bit.offset}); + } + result.push_back(BridgeColumnInfo{rust::String(info.name), info.type, info.decodedSize, std::move(bitfield)}); + } + return result; +} + +rust::Vec FrameWrapper::properties() const { + rust::Vec result; + for (const auto& [key, value] : frame_.properties()) { + result.push_back(BridgeProperty{rust::String(key), rust::String(value)}); + } + return result; +} + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/FrameWrapper.h b/rust/crates/odc-sys/cpp/FrameWrapper.h new file mode 100644 index 00000000..d2357ad7 --- /dev/null +++ b/rust/crates/odc-sys/cpp/FrameWrapper.h @@ -0,0 +1,39 @@ +// odc Frame bridge — wraps `odc::api::Frame`. +#pragma once + +#include "odc/api/Odb.h" + +#include "rust/cxx.h" + +#include + +namespace odc_bridge { + +// cxx-generated shared structs — defined in "odc-sys/src/lib.rs.h", which +// the .cc includes. Forward-declared here to avoid a circular include (the +// generated header includes this one first). +struct BridgeBit; +struct BridgeColumnInfo; +struct BridgeProperty; + +/// Owns an `odc::api::Frame` — a viewport onto a chunk of contiguous, +/// compatible data (possibly a logical frame aggregating several physical +/// frames). The frame reads lazily from the Reader's stream, so it must not +/// outlive its ReaderWrapper — enforced on the Rust side. +class FrameWrapper { + odc::api::Frame frame_; + + friend class DecoderWrapper; + +public: + + explicit FrameWrapper(odc::api::Frame&& frame); + + size_t row_count() const; + size_t column_count() const; + bool has_column(rust::Str name) const; + rust::Vec column_info() const; + rust::Vec properties() const; +}; + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/OdcBridge.cc b/rust/crates/odc-sys/cpp/OdcBridge.cc deleted file mode 100644 index 14212a59..00000000 --- a/rust/crates/odc-sys/cpp/OdcBridge.cc +++ /dev/null @@ -1,148 +0,0 @@ -// odc C++ bridge implementation -#include "OdcBridge.h" - -#include "odc-sys/src/lib.rs.h" - -#include "eckit/exception/Exceptions.h" - -#include - -namespace odc_bridge { - -// ==================== FrameWrapper ==================== - -FrameWrapper::FrameWrapper(odc::api::Frame&& frame) : frame_(std::move(frame)) {} - -size_t FrameWrapper::row_count() const { - return frame_.rowCount(); -} - -size_t FrameWrapper::column_count() const { - return frame_.columnCount(); -} - -bool FrameWrapper::has_column(rust::Str name) const { - return frame_.hasColumn(std::string(name)); -} - -rust::Vec FrameWrapper::column_info() const { - rust::Vec result; - result.reserve(frame_.columnCount()); - for (const auto& info : frame_.columnInfo()) { - rust::Vec bitfield; - bitfield.reserve(info.bitfield.size()); - for (const auto& bit : info.bitfield) { - bitfield.push_back(BridgeBit{rust::String(bit.name), bit.size, bit.offset}); - } - result.push_back( - BridgeColumnInfo{rust::String(info.name), info.type, info.decodedSize, std::move(bitfield)}); - } - return result; -} - -rust::Vec FrameWrapper::properties() const { - rust::Vec result; - for (const auto& [key, value] : frame_.properties()) { - result.push_back(BridgeProperty{rust::String(key), rust::String(value)}); - } - return result; -} - -// ==================== ReaderWrapper ==================== - -ReaderWrapper::ReaderWrapper(std::unique_ptr reader) : reader_(std::move(reader)) {} - -std::unique_ptr ReaderWrapper::from_path(rust::Str path, bool aggregated, int64_t rowlimit) { - return std::make_unique( - std::make_unique(std::string(path), aggregated, static_cast(rowlimit))); -} - -std::unique_ptr ReaderWrapper::from_handle(eckit_bridge::DataHandleWrapper& handle, bool aggregated, - int64_t rowlimit) { - return std::make_unique( - std::make_unique(handle.inner(), aggregated, static_cast(rowlimit))); -} - -std::unique_ptr ReaderWrapper::next_frame() { - odc::api::Frame frame = reader_->next(); - if (!frame) { - return nullptr; - } - return std::make_unique(std::move(frame)); -} - -// ==================== DecoderWrapper ==================== - -std::unique_ptr DecoderWrapper::create() { - return std::make_unique(); -} - -void DecoderWrapper::add_column(rust::Str name, uint8_t* data, size_t nrows, size_t elem_size, size_t stride) { - names_.emplace_back(std::string(name)); - facades_.emplace_back(reinterpret_cast(data), nrows, elem_size, stride); -} - -size_t DecoderWrapper::decode(const FrameWrapper& frame, size_t nthreads) { - odc::api::Decoder decoder(names_, facades_); - decoder.decode(frame.frame_, nthreads == 0 ? 1 : nthreads); - return frame.frame_.rowCount(); -} - -// ==================== EncoderWrapper ==================== - -std::unique_ptr EncoderWrapper::create() { - return std::make_unique(); -} - -void EncoderWrapper::add_column(rust::Str name, ColumnType column_type, size_t elem_size, const uint8_t* data, - size_t nrows, size_t stride) { - columns_.push_back(odc::api::ColumnInfo{std::string(name), column_type, elem_size, {}}); - data_.emplace_back(reinterpret_cast(data), nrows, elem_size, stride); -} - -void EncoderWrapper::add_bitfield(rust::Str name, int32_t size, int32_t offset) { - if (columns_.empty()) { - throw eckit::UserError("add_bitfield called before add_column"); - } - columns_.back().bitfield.push_back(odc::api::ColumnInfo::Bit{std::string(name), size, offset}); -} - -void EncoderWrapper::set_property(rust::Str key, rust::Str value) { - properties_[std::string(key)] = std::string(value); -} - -void EncoderWrapper::encode(eckit_bridge::DataHandleWrapper& out, size_t max_rows_per_frame) { - odc::api::encode(out.inner(), columns_, data_, properties_, max_rows_per_frame); -} - -// ==================== SettingsWrapper ==================== - -void SettingsWrapper::treat_integers_as_doubles(bool flag) { - odc::api::Settings::treatIntegersAsDoubles(flag); -} - -int64_t SettingsWrapper::integer_missing_value() { - return static_cast(odc::api::Settings::integerMissingValue()); -} - -void SettingsWrapper::set_integer_missing_value(int64_t value) { - odc::api::Settings::setIntegerMissingValue(static_cast(value)); -} - -double SettingsWrapper::double_missing_value() { - return odc::api::Settings::doubleMissingValue(); -} - -void SettingsWrapper::set_double_missing_value(double value) { - odc::api::Settings::setDoubleMissingValue(value); -} - -rust::String SettingsWrapper::version() { - return rust::String(odc::api::Settings::version()); -} - -rust::String SettingsWrapper::gitsha1() { - return rust::String(odc::api::Settings::gitsha1()); -} - -} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/OdcBridge.h b/rust/crates/odc-sys/cpp/OdcBridge.h index 5936e5bc..0338d14b 100644 --- a/rust/crates/odc-sys/cpp/OdcBridge.h +++ b/rust/crates/odc-sys/cpp/OdcBridge.h @@ -1,139 +1,17 @@ -// odc C++ bridge for Rust FFI +// odc C++ bridge for Rust FFI — umbrella header pulled in by the +// cxx-generated bridge (`include!("OdcBridge.h")` in lib.rs). Real +// declarations live in the per-topic headers below. #pragma once -#include "EckitBridge.h" -#include "odc_exceptions.h" - -#include "odc/api/ColumnType.h" -#include "odc/api/Odb.h" - -#include "rust/cxx.h" - -#include -#include -#include -#include -#include - -namespace odc_bridge { - -// ColumnType is odc::api::ColumnType — cxx static_asserts values match. -using odc::api::ColumnType; - -// cxx-generated shared structs — defined in "odc-sys/src/lib.rs.h", which -// OdcBridge.cc includes. Forward-declared here to avoid a circular include -// (the generated header includes OdcBridge.h first). -struct BridgeBit; -struct BridgeColumnInfo; -struct BridgeProperty; - -// ==================== FrameWrapper ==================== - -/// Owns an `odc::api::Frame` — a viewport onto a chunk of contiguous, -/// compatible data (possibly a logical frame aggregating several physical -/// frames). The frame reads lazily from the Reader's stream, so it must not -/// outlive its ReaderWrapper — enforced on the Rust side. -class FrameWrapper { - odc::api::Frame frame_; - - friend class DecoderWrapper; - -public: - - explicit FrameWrapper(odc::api::Frame&& frame); - - size_t row_count() const; - size_t column_count() const; - bool has_column(rust::Str name) const; - rust::Vec column_info() const; - rust::Vec properties() const; -}; - -// ==================== ReaderWrapper ==================== - -/// Wraps `odc::api::Reader` — owns the ODB-2 stream, yields frames. -class ReaderWrapper { - std::unique_ptr reader_; - -public: - - explicit ReaderWrapper(std::unique_ptr reader); - - static std::unique_ptr from_path(rust::Str path, bool aggregated, int64_t rowlimit); - - /// Does not take ownership: the DataHandleWrapper must outlive this - /// reader (and every frame it yields) — enforced on the Rust side. - /// The handle must not be open; the reader opens it for reading. - static std::unique_ptr from_handle(eckit_bridge::DataHandleWrapper& handle, bool aggregated, - int64_t rowlimit); - - /// Returns nullptr when the stream is exhausted. - std::unique_ptr next_frame(); -}; - -// ==================== DecoderWrapper ==================== - -/// Accumulates per-column decode targets (StridedData facades over -/// Rust-owned buffers), then runs `odc::api::Decoder::decode`. -class DecoderWrapper { - std::vector names_; - std::vector facades_; - -public: - - static std::unique_ptr create(); - - /// `data` must point to caller-owned, 8-byte-aligned memory of at least - /// nrows * stride bytes, valid until decode() returns. - void add_column(rust::Str name, uint8_t* data, size_t nrows, size_t elem_size, size_t stride); - - /// Decode the frame into the registered buffers; returns rows decoded. - size_t decode(const FrameWrapper& frame, size_t nthreads); -}; - -// ==================== EncoderWrapper ==================== - -/// Accumulates column specs + ConstStridedData facades over Rust-owned -/// buffers, then calls the `odc::api::encode` free function. -class EncoderWrapper { - std::vector columns_; - std::vector data_; - std::map properties_; - -public: - - static std::unique_ptr create(); - - /// `data` must point to caller-owned memory of at least nrows * stride - /// bytes, valid until encode() returns. - void add_column(rust::Str name, ColumnType column_type, size_t elem_size, const uint8_t* data, size_t nrows, - size_t stride); - - /// Append a bit group to the most recently added column. - void add_bitfield(rust::Str name, int32_t size, int32_t offset); - - void set_property(rust::Str key, rust::Str value); - - /// Encode all columns to an (already open) data handle. - void encode(eckit_bridge::DataHandleWrapper& out, size_t max_rows_per_frame); -}; - -// ==================== SettingsWrapper ==================== - -/// Wraps `odc::api::Settings` — process-global settings and version info. -/// Static-only; never instantiated. -class SettingsWrapper { -public: - - SettingsWrapper() = delete; - - static void treat_integers_as_doubles(bool flag); - static int64_t integer_missing_value(); - static void set_integer_missing_value(int64_t value); - static double double_missing_value(); - static void set_double_missing_value(double value); - static rust::String version(); - static rust::String gitsha1(); -}; - -} // namespace odc_bridge +// Note: the auto-generated `rust::behavior::trycatch` lives in +// `odc_exceptions.h`, which lib.rs pulls into the cxx-generated translation +// unit via its own `include!` (before this header). It must not be included +// from here: downstream `-sys` crates have their own generated +// `_exceptions.h` and must not see odc's transitively, or they would +// have two `trycatch` specializations in one translation unit. + +#include "DecoderWrapper.h" +#include "EncoderWrapper.h" +#include "FrameWrapper.h" +#include "ReaderWrapper.h" +#include "SettingsWrapper.h" diff --git a/rust/crates/odc-sys/cpp/ReaderWrapper.cc b/rust/crates/odc-sys/cpp/ReaderWrapper.cc new file mode 100644 index 00000000..b1152afa --- /dev/null +++ b/rust/crates/odc-sys/cpp/ReaderWrapper.cc @@ -0,0 +1,32 @@ +// odc Reader bridge — implementation. + +#include "ReaderWrapper.h" +#include "odc-sys/src/lib.rs.h" + +#include +#include + +namespace odc_bridge { + +ReaderWrapper::ReaderWrapper(std::unique_ptr reader) : reader_(std::move(reader)) {} + +std::unique_ptr ReaderWrapper::from_path(rust::Str path, bool aggregated, int64_t rowlimit) { + return std::make_unique( + std::make_unique(std::string(path), aggregated, static_cast(rowlimit))); +} + +std::unique_ptr ReaderWrapper::from_handle(eckit_bridge::DataHandleWrapper& handle, bool aggregated, + int64_t rowlimit) { + return std::make_unique( + std::make_unique(handle.inner(), aggregated, static_cast(rowlimit))); +} + +std::unique_ptr ReaderWrapper::next_frame() { + odc::api::Frame frame = reader_->next(); + if (!frame) { + return nullptr; + } + return std::make_unique(std::move(frame)); +} + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/ReaderWrapper.h b/rust/crates/odc-sys/cpp/ReaderWrapper.h new file mode 100644 index 00000000..31085d4a --- /dev/null +++ b/rust/crates/odc-sys/cpp/ReaderWrapper.h @@ -0,0 +1,36 @@ +// odc Reader bridge — wraps `odc::api::Reader`. +#pragma once + +#include "DataHandleWrapper.h" +#include "FrameWrapper.h" + +#include "odc/api/Odb.h" + +#include "rust/cxx.h" + +#include +#include + +namespace odc_bridge { + +/// Wraps `odc::api::Reader` — owns the ODB-2 stream, yields frames. +class ReaderWrapper { + std::unique_ptr reader_; + +public: + + explicit ReaderWrapper(std::unique_ptr reader); + + static std::unique_ptr from_path(rust::Str path, bool aggregated, int64_t rowlimit); + + /// Does not take ownership: the DataHandleWrapper must outlive this + /// reader (and every frame it yields) — enforced on the Rust side. + /// The handle must not be open; the reader opens it for reading. + static std::unique_ptr from_handle(eckit_bridge::DataHandleWrapper& handle, bool aggregated, + int64_t rowlimit); + + /// Returns nullptr when the stream is exhausted. + std::unique_ptr next_frame(); +}; + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/SettingsWrapper.cc b/rust/crates/odc-sys/cpp/SettingsWrapper.cc new file mode 100644 index 00000000..92a0f2f4 --- /dev/null +++ b/rust/crates/odc-sys/cpp/SettingsWrapper.cc @@ -0,0 +1,38 @@ +// odc Settings bridge — implementation. + +#include "SettingsWrapper.h" +#include "odc-sys/src/lib.rs.h" + +#include "odc/api/Odb.h" + +namespace odc_bridge { + +void SettingsWrapper::treat_integers_as_doubles(bool flag) { + odc::api::Settings::treatIntegersAsDoubles(flag); +} + +int64_t SettingsWrapper::integer_missing_value() { + return static_cast(odc::api::Settings::integerMissingValue()); +} + +void SettingsWrapper::set_integer_missing_value(int64_t value) { + odc::api::Settings::setIntegerMissingValue(static_cast(value)); +} + +double SettingsWrapper::double_missing_value() { + return odc::api::Settings::doubleMissingValue(); +} + +void SettingsWrapper::set_double_missing_value(double value) { + odc::api::Settings::setDoubleMissingValue(value); +} + +rust::String SettingsWrapper::version() { + return rust::String(odc::api::Settings::version()); +} + +rust::String SettingsWrapper::gitsha1() { + return rust::String(odc::api::Settings::gitsha1()); +} + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/cpp/SettingsWrapper.h b/rust/crates/odc-sys/cpp/SettingsWrapper.h new file mode 100644 index 00000000..02c91413 --- /dev/null +++ b/rust/crates/odc-sys/cpp/SettingsWrapper.h @@ -0,0 +1,26 @@ +// odc Settings bridge — wraps `odc::api::Settings`. +#pragma once + +#include "rust/cxx.h" + +#include + +namespace odc_bridge { + +/// Wraps `odc::api::Settings` — process-global settings and version info. +/// Static-only; never instantiated. +class SettingsWrapper { +public: + + SettingsWrapper() = delete; + + static void treat_integers_as_doubles(bool flag); + static int64_t integer_missing_value(); + static void set_integer_missing_value(int64_t value); + static double double_missing_value(); + static void set_double_missing_value(double value); + static rust::String version(); + static rust::String gitsha1(); +}; + +} // namespace odc_bridge diff --git a/rust/crates/odc-sys/src/lib.rs b/rust/crates/odc-sys/src/lib.rs index b837b45f..765a53a1 100644 --- a/rust/crates/odc-sys/src/lib.rs +++ b/rust/crates/odc-sys/src/lib.rs @@ -64,6 +64,10 @@ pub mod ffi { } unsafe extern "C++" { + // odc_exceptions.h first: it defines the `rust::behavior::trycatch` + // that maps C++ exceptions to typed errors, and must be visible in + // the cxx-generated translation unit before the wrapper headers. + include!("odc_exceptions.h"); include!("OdcBridge.h"); // Verify ColumnType matches C++ odc::api::ColumnType at compile time From 5a46e9bae4f26468f1bdb3e4db276ef3025bb17a Mon Sep 17 00:00:00 2001 From: Vlad Pankratov Date: Thu, 16 Jul 2026 20:28:45 +0200 Subject: [PATCH 7/7] Update workspace configuration to include new `odc` crate and adjust dependencies --- {.cargo => rust/.cargo}/config.toml | 0 rust/Cargo.toml | 8 +- rust/crates/odc-sys/cpp/FrameWrapper.cc | 16 +- rust/crates/odc-sys/cpp/FrameWrapper.h | 10 +- rust/crates/odc-sys/src/lib.rs | 26 ++- rust/crates/odc/Cargo.toml | 27 +++ rust/crates/odc/README.md | 18 ++ rust/crates/odc/src/decode.rs | 156 ++++++++++++++++ rust/crates/odc/src/encode.rs | 237 ++++++++++++++++++++++++ rust/crates/odc/src/error.rs | 72 +++++++ rust/crates/odc/src/frame.rs | 125 +++++++++++++ rust/crates/odc/src/lib.rs | 152 +++++++++++++++ rust/crates/odc/src/reader.rs | 158 ++++++++++++++++ 13 files changed, 982 insertions(+), 23 deletions(-) rename {.cargo => rust/.cargo}/config.toml (100%) create mode 100644 rust/crates/odc/Cargo.toml create mode 100644 rust/crates/odc/README.md create mode 100644 rust/crates/odc/src/decode.rs create mode 100644 rust/crates/odc/src/encode.rs create mode 100644 rust/crates/odc/src/error.rs create mode 100644 rust/crates/odc/src/frame.rs create mode 100644 rust/crates/odc/src/lib.rs create mode 100644 rust/crates/odc/src/reader.rs diff --git a/.cargo/config.toml b/rust/.cargo/config.toml similarity index 100% rename from .cargo/config.toml rename to rust/.cargo/config.toml diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 3b5881b6..5221023f 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -1,6 +1,6 @@ [workspace] resolver = "2" -members = ["crates/odc-sys"] +members = ["crates/odc", "crates/odc-sys"] [workspace.package] edition = "2024" @@ -13,9 +13,11 @@ categories = ["science"] [workspace.dependencies] # Internal -odc-sys = { path = "crates/odc-sys" } +odc = { path = "crates/odc", default-features = false } +odc-sys = { path = "crates/odc-sys", default-features = false } # Foundation crates +eckit = { git = "ssh://git@github.com/ecmwf/rust-wrappers-playground.git", default-features = false } eckit-sys = { git = "ssh://git@github.com/ecmwf/eckit.git", branch = "rust-bindings", default-features = false } # Build tools @@ -26,4 +28,6 @@ bindman-utils = { git = "ssh://git@github.com/ecmwf/bindman.git", rev = "47edf68 # External cxx = "1.0" cxx-build = "1.0" +parking_lot = "0.12" +polars = { version = "0.54", default-features = false, features = ["fmt"] } thiserror = "2" diff --git a/rust/crates/odc-sys/cpp/FrameWrapper.cc b/rust/crates/odc-sys/cpp/FrameWrapper.cc index 4d494a4e..480e2bb1 100644 --- a/rust/crates/odc-sys/cpp/FrameWrapper.cc +++ b/rust/crates/odc-sys/cpp/FrameWrapper.cc @@ -22,24 +22,24 @@ bool FrameWrapper::has_column(rust::Str name) const { return frame_.hasColumn(std::string(name)); } -rust::Vec FrameWrapper::column_info() const { - rust::Vec result; +rust::Vec FrameWrapper::column_info() const { + rust::Vec result; result.reserve(frame_.columnCount()); for (const auto& info : frame_.columnInfo()) { - rust::Vec bitfield; + rust::Vec bitfield; bitfield.reserve(info.bitfield.size()); for (const auto& bit : info.bitfield) { - bitfield.push_back(BridgeBit{rust::String(bit.name), bit.size, bit.offset}); + bitfield.push_back(Bit{rust::String(bit.name), bit.size, bit.offset}); } - result.push_back(BridgeColumnInfo{rust::String(info.name), info.type, info.decodedSize, std::move(bitfield)}); + result.push_back(ColumnInfo{rust::String(info.name), info.type, info.decodedSize, std::move(bitfield)}); } return result; } -rust::Vec FrameWrapper::properties() const { - rust::Vec result; +rust::Vec FrameWrapper::properties() const { + rust::Vec result; for (const auto& [key, value] : frame_.properties()) { - result.push_back(BridgeProperty{rust::String(key), rust::String(value)}); + result.push_back(Property{rust::String(key), rust::String(value)}); } return result; } diff --git a/rust/crates/odc-sys/cpp/FrameWrapper.h b/rust/crates/odc-sys/cpp/FrameWrapper.h index d2357ad7..bf8bc159 100644 --- a/rust/crates/odc-sys/cpp/FrameWrapper.h +++ b/rust/crates/odc-sys/cpp/FrameWrapper.h @@ -12,9 +12,9 @@ namespace odc_bridge { // cxx-generated shared structs — defined in "odc-sys/src/lib.rs.h", which // the .cc includes. Forward-declared here to avoid a circular include (the // generated header includes this one first). -struct BridgeBit; -struct BridgeColumnInfo; -struct BridgeProperty; +struct Bit; +struct ColumnInfo; +struct Property; /// Owns an `odc::api::Frame` — a viewport onto a chunk of contiguous, /// compatible data (possibly a logical frame aggregating several physical @@ -32,8 +32,8 @@ class FrameWrapper { size_t row_count() const; size_t column_count() const; bool has_column(rust::Str name) const; - rust::Vec column_info() const; - rust::Vec properties() const; + rust::Vec column_info() const; + rust::Vec properties() const; }; } // namespace odc_bridge diff --git a/rust/crates/odc-sys/src/lib.rs b/rust/crates/odc-sys/src/lib.rs index 765a53a1..1e483055 100644 --- a/rust/crates/odc-sys/src/lib.rs +++ b/rust/crates/odc-sys/src/lib.rs @@ -8,6 +8,7 @@ use bindman::track_cpp_api; // Auto-generated odc Error enum + From impl include!(concat!(env!("OUT_DIR"), "/odc_exceptions.rs")); +#[allow(clippy::missing_safety_doc)] #[track_cpp_api( ("odc/api/Odb.h", class = "Reader"), ("odc/api/Odb.h", class = "Frame"), @@ -35,9 +36,9 @@ pub mod ffi { Double = 5, } - /// A bit group within a bitfield column. + /// A bit group within a bitfield column — C++ `odc::api::ColumnInfo::Bit`. #[derive(Debug, Clone, PartialEq, Eq)] - struct BridgeBit { + struct Bit { name: String, /// Bit group size in bits. size: i32, @@ -45,20 +46,20 @@ pub mod ffi { offset: i32, } - /// Metadata for one column of a frame. + /// Metadata for one column of a frame — C++ `odc::api::ColumnInfo`. #[derive(Debug, Clone)] - struct BridgeColumnInfo { + struct ColumnInfo { name: String, column_type: ColumnType, /// Size of a single decoded value in bytes (always a multiple of 8). decoded_size: usize, /// Bit groups — non-empty only for bitfield columns. - bitfield: Vec, + bitfield: Vec, } /// A key/value property encoded in a frame. #[derive(Debug, Clone, PartialEq, Eq)] - struct BridgeProperty { + struct Property { key: String, value: String, } @@ -107,17 +108,21 @@ pub mod ffi { type FrameWrapper; + #[must_use] fn row_count(self: &FrameWrapper) -> usize; + #[must_use] fn column_count(self: &FrameWrapper) -> usize; + #[must_use] fn has_column(self: &FrameWrapper, name: &str) -> bool; - fn column_info(self: &FrameWrapper) -> Result>; - fn properties(self: &FrameWrapper) -> Result>; + fn column_info(self: &FrameWrapper) -> Result>; + fn properties(self: &FrameWrapper) -> Result>; // ==================== DecoderWrapper ==================== type DecoderWrapper; #[Self = "DecoderWrapper"] + #[must_use] fn create() -> UniquePtr; /// Register a decode target for the named column. @@ -147,6 +152,7 @@ pub mod ffi { type EncoderWrapper; #[Self = "EncoderWrapper"] + #[must_use] fn create() -> UniquePtr; /// Register a source column for encoding. @@ -191,16 +197,20 @@ pub mod ffi { #[Self = "SettingsWrapper"] fn treat_integers_as_doubles(flag: bool); #[Self = "SettingsWrapper"] + #[must_use] fn integer_missing_value() -> i64; #[Self = "SettingsWrapper"] fn set_integer_missing_value(value: i64); #[Self = "SettingsWrapper"] + #[must_use] fn double_missing_value() -> f64; #[Self = "SettingsWrapper"] fn set_double_missing_value(value: f64); #[Self = "SettingsWrapper"] + #[must_use] fn version() -> String; #[Self = "SettingsWrapper"] + #[must_use] fn gitsha1() -> String; } } diff --git a/rust/crates/odc/Cargo.toml b/rust/crates/odc/Cargo.toml new file mode 100644 index 00000000..080a8ffd --- /dev/null +++ b/rust/crates/odc/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "odc" +version = "0.1.0" +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +readme = "README.md" +keywords.workspace = true +categories.workspace = true +description = "Safe Rust wrapper for ECMWF's odc (ODB-2 encoder/decoder) library" + +[features] +default = ["vendored"] + +# Build strategy (mutually exclusive) +vendored = ["odc-sys/vendored", "eckit/vendored"] +system = ["odc-sys/system", "eckit/system"] + +[dependencies] +odc-sys.workspace = true +eckit.workspace = true +parking_lot.workspace = true +polars.workspace = true +thiserror.workspace = true + +[package.metadata.docs.rs] diff --git a/rust/crates/odc/README.md b/rust/crates/odc/README.md new file mode 100644 index 00000000..1798b59c --- /dev/null +++ b/rust/crates/odc/README.md @@ -0,0 +1,18 @@ +# odc + +Safe Rust wrapper for ECMWF's [odc](https://github.com/ecmwf/odc) (ODB-2 encoder/decoder) C++ library. + +ODB-2 data decodes into [Polars](https://pola.rs/) data frames and encodes from them, following the same model as [pyodc](https://github.com/ecmwf/pyodc). For raw FFI bindings, see the lower-level `odc-sys` crate. + +## Features + +### Build strategy (mutually exclusive) + +- `vendored` - Build odc and its dependencies (eckit) from source. +- `system` - Link against system-installed odc. + +`vendored` is enabled by default. + +## License + +Apache-2.0 diff --git a/rust/crates/odc/src/decode.rs b/rust/crates/odc/src/decode.rs new file mode 100644 index 00000000..2e5f7eb1 --- /dev/null +++ b/rust/crates/odc/src/decode.rs @@ -0,0 +1,156 @@ +//! Decoding a [`Frame`] into a Polars `DataFrame`. + +use odc_sys::{ColumnInfo, ColumnType, SettingsWrapper}; +use polars::prelude::*; + +use crate::error::{Error, Result}; +use crate::frame::{DecodeOptions, Frame}; + +/// Decode target for one column. All slots are 8 bytes: with +/// integers-as-longs behaviour, INTEGER/BITFIELD decode as `i64`, REAL and +/// DOUBLE as `f64`, and STRING as fixed-width byte blocks (`Vec` keeps +/// them 8-byte aligned). +enum Buffer { + I64(Vec), + F64(Vec), + Str { data: Vec, width: usize }, +} + +impl Buffer { + fn for_column(col: &ColumnInfo, nrows: usize) -> Self { + match col.column_type { + ColumnType::Integer | ColumnType::Bitfield => Self::I64(vec![0; nrows]), + ColumnType::String => { + let width = col.decoded_size.max(8); + Self::Str { + data: vec![0; nrows * width / 8], + width, + } + } + _ => Self::F64(vec![0.0; nrows]), + } + } + + const fn as_mut_ptr(&mut self) -> *mut u8 { + match self { + Self::I64(v) => v.as_mut_ptr().cast(), + Self::F64(v) => v.as_mut_ptr().cast(), + Self::Str { data, .. } => data.as_mut_ptr().cast(), + } + } + + const fn elem_size(&self) -> usize { + match self { + Self::I64(_) | Self::F64(_) => 8, + Self::Str { width, .. } => *width, + } + } +} + +pub fn dataframe(frame: &Frame, options: &DecodeOptions) -> Result { + let nrows = frame.row_count(); + + let selected: Vec<&ColumnInfo> = match &options.columns { + Some(names) => names + .iter() + .map(|name| { + frame + .column(name) + .ok_or_else(|| Error::ColumnNotFound(name.clone())) + }) + .collect::>()?, + None => frame + .columns() + .iter() + .filter(|c| c.column_type != ColumnType::Ignore) + .collect(), + }; + if let Some(col) = selected + .iter() + .find(|c| c.column_type == ColumnType::Ignore) + { + return Err(Error::UnsupportedColumnType { + column: col.name.clone(), + column_type: col.column_type, + }); + } + + // All buffers are allocated up front so no Vec reallocation can move + // them while the decoder holds raw pointers into them. + let mut buffers: Vec = selected + .iter() + .map(|col| Buffer::for_column(col, nrows)) + .collect(); + + let mut decoder = odc_sys::DecoderWrapper::create(); + for (col, buf) in selected.iter().zip(&mut buffers) { + let elem_size = buf.elem_size(); + // SAFETY: each buffer is 8-byte aligned (Vec/Vec/Vec), + // holds nrows * elem_size bytes, and outlives the decode call below. + unsafe { + decoder + .pin_mut() + .add_column(&col.name, buf.as_mut_ptr(), nrows, elem_size, elem_size); + } + } + decoder.pin_mut().decode(frame.wrapper(), options.threads)?; + + let columns = selected + .iter() + .zip(buffers) + .map(|(col, buf)| to_polars(col, buf, nrows)) + .collect::>>()?; + Ok(DataFrame::new(nrows, columns)?) +} + +fn to_polars(col: &ColumnInfo, buffer: Buffer, nrows: usize) -> Result { + let name: PlSmallStr = col.name.as_str().into(); + let series = match buffer { + // Bitfields are raw bit patterns — no missing-value mapping. + Buffer::I64(values) if col.column_type == ColumnType::Bitfield => Series::new(name, values), + Buffer::I64(values) => { + let missing = SettingsWrapper::integer_missing_value(); + if values.contains(&missing) { + let values: Vec> = values + .iter() + .map(|&v| (v != missing).then_some(v)) + .collect(); + Series::new(name, values) + } else { + Series::new(name, values) + } + } + Buffer::F64(values) => { + let missing = SettingsWrapper::double_missing_value().to_bits(); + let series = if values.iter().any(|v| v.to_bits() == missing) { + let values: Vec> = values + .iter() + .map(|&v| (v.to_bits() != missing).then_some(v)) + .collect(); + Series::new(name, values) + } else { + Series::new(name, values) + }; + if col.column_type == ColumnType::Real { + series.cast(&DataType::Float32)? + } else { + series + } + } + Buffer::Str { data, width } => { + let slots = width / 8; + let strings: Vec = (0..nrows) + .map(|row| { + let cell: Vec = data[row * slots..(row + 1) * slots] + .iter() + .flat_map(|slot| slot.to_ne_bytes()) + .collect(); + let end = cell.iter().rposition(|&b| b != 0).map_or(0, |i| i + 1); + String::from_utf8_lossy(&cell[..end]).into_owned() + }) + .collect(); + Series::new(name, strings) + } + }; + Ok(series.into_column()) +} diff --git a/rust/crates/odc/src/encode.rs b/rust/crates/odc/src/encode.rs new file mode 100644 index 00000000..fed437c7 --- /dev/null +++ b/rust/crates/odc/src/encode.rs @@ -0,0 +1,237 @@ +//! Encoding a Polars `DataFrame` into ODB-2. + +use std::collections::{BTreeMap, HashMap}; +use std::path::Path; + +use odc_sys::{Bit, ColumnType, SettingsWrapper}; +use polars::prelude::*; + +use crate::error::{Error, Result}; +use crate::init; + +/// Options for [`write_odb`]. +#[derive(Debug, Clone)] +pub struct WriteOptions { + /// Maximum number of rows per physical output frame. + pub rows_per_frame: usize, + /// Per-column overrides of the dtype-derived ODB column type. Supported: + /// `Integer` → `Bitfield` (requires a [`bitfields`](Self::bitfields) + /// entry) and `Double` ↔ `Real`. + pub types: HashMap, + /// Key/value properties attached to every output frame. + pub properties: BTreeMap, + /// Bit group layout for columns encoded as `Bitfield`. + pub bitfields: HashMap>, +} + +impl Default for WriteOptions { + fn default() -> Self { + Self { + rows_per_frame: 10_000, + types: HashMap::new(), + properties: BTreeMap::new(), + bitfields: HashMap::new(), + } + } +} + +/// Encode a `DataFrame` into an ODB-2 file. +/// +/// Column types derive from dtypes: `Int64` (and smaller integers / +/// `Boolean`, widened) → INTEGER, `Float64` → DOUBLE, `Float32` → REAL, +/// `String` → STRING; nulls become ODB missing values. Other dtypes are +/// rejected. +/// +/// # Errors +/// +/// Fails on an empty `DataFrame`, unsupported dtypes, invalid type +/// overrides or bitfield specifications, or if the file cannot be written. +pub fn write_odb(df: &DataFrame, path: impl AsRef, options: &WriteOptions) -> Result<()> { + init(); + let handle = eckit::DataHandle::from_path(path)?; + let mut handle = handle.open_for_write(0)?; + let result = write_odb_to(df, &mut handle, options); + let closed = handle.close(); + result?; + closed?; + Ok(()) +} + +/// Encode a `DataFrame` into an open eckit +/// [`DataHandle`](eckit::DataHandle) (file, buffer, tee, …). +/// +/// # Errors +/// +/// See [`write_odb`]. +pub fn write_odb_to( + df: &DataFrame, + handle: &mut eckit::DataHandle, + options: &WriteOptions, +) -> Result<()> { + init(); + let nrows = df.height(); + if nrows == 0 || df.width() == 0 { + return Err(Error::EmptyDataFrame); + } + + let missing_int = SettingsWrapper::integer_missing_value(); + let missing_dbl = SettingsWrapper::double_missing_value(); + + let mut staged: Vec<(String, ColumnType, Staged)> = Vec::with_capacity(df.width()); + for column in df.columns() { + let name = column.name().to_string(); + let series = column.as_materialized_series().rechunk(); + + let (natural, data) = match series.dtype() { + DataType::Int64 => (ColumnType::Integer, stage_i64(&series, missing_int)?), + DataType::Int8 + | DataType::Int16 + | DataType::Int32 + | DataType::UInt8 + | DataType::UInt16 + | DataType::UInt32 + | DataType::Boolean => ( + ColumnType::Integer, + stage_i64(&series.cast(&DataType::Int64)?, missing_int)?, + ), + DataType::Float64 => (ColumnType::Double, stage_f64(&series, missing_dbl)?), + DataType::Float32 => ( + ColumnType::Real, + stage_f64(&series.cast(&DataType::Float64)?, missing_dbl)?, + ), + DataType::String => (ColumnType::String, stage_str(&series)), + other => { + return Err(Error::UnsupportedDtype { + column: name, + dtype: other.to_string(), + }); + } + }; + + let target = options.types.get(&name).copied().unwrap_or(natural); + let compatible = target == natural + || (natural == ColumnType::Integer && target == ColumnType::Bitfield) + || (natural == ColumnType::Double && target == ColumnType::Real) + || (natural == ColumnType::Real && target == ColumnType::Double); + if !compatible { + return Err(Error::InvalidTypeOverride { + column: name, + from: natural, + to: target, + }); + } + if target == ColumnType::Bitfield { + validate_bitfield(&name, options.bitfields.get(&name))?; + } + + staged.push((name, target, data)); + } + + let mut encoder = odc_sys::EncoderWrapper::create(); + for (name, column_type, data) in &staged { + let elem_size = data.elem_size(); + // SAFETY: the staged buffers live in `staged` until after the + // encode call below. + unsafe { + encoder.pin_mut().add_column( + name, + *column_type, + elem_size, + data.as_ptr(), + nrows, + elem_size, + ); + } + if *column_type == ColumnType::Bitfield + && let Some(bits) = options.bitfields.get(name) + { + for bit in bits { + encoder + .pin_mut() + .add_bitfield(&bit.name, bit.size, bit.offset)?; + } + } + } + for (key, value) in &options.properties { + encoder.pin_mut().set_property(key, value); + } + encoder + .pin_mut() + .encode(handle.inner_mut()?, options.rows_per_frame)?; + Ok(()) +} + +/// Staged (contiguous, null-resolved) source data for one column. +enum Staged { + I64(Vec), + F64(Vec), + Bytes { data: Vec, width: usize }, +} + +impl Staged { + const fn as_ptr(&self) -> *const u8 { + match self { + Self::I64(v) => v.as_ptr().cast(), + Self::F64(v) => v.as_ptr().cast(), + Self::Bytes { data, .. } => data.as_ptr(), + } + } + + const fn elem_size(&self) -> usize { + match self { + Self::I64(_) | Self::F64(_) => 8, + Self::Bytes { width, .. } => *width, + } + } +} + +fn stage_i64(series: &Series, missing: i64) -> Result { + let values = series.i64()?; + Ok(Staged::I64( + values.iter().map(|v| v.unwrap_or(missing)).collect(), + )) +} + +fn stage_f64(series: &Series, missing: f64) -> Result { + let values = series.f64()?; + Ok(Staged::F64( + values.iter().map(|v| v.unwrap_or(missing)).collect(), + )) +} + +fn stage_str(series: &Series) -> Staged { + // Fixed width: longest value rounded up to a multiple of 8 (min 8), + // NUL-padded. Nulls encode as the empty string. + let values: Vec> = series + .str() + .map_or_else(|_| Vec::new(), |ca| ca.iter().collect()); + let longest = values + .iter() + .map(|v| v.map_or(0, str::len)) + .max() + .unwrap_or(0); + let width = longest.max(1).div_ceil(8) * 8; + let mut data = vec![0_u8; series.len() * width]; + for (row, value) in values.iter().enumerate() { + if let Some(value) = value { + data[row * width..row * width + value.len()].copy_from_slice(value.as_bytes()); + } + } + Staged::Bytes { data, width } +} + +fn validate_bitfield(column: &str, bits: Option<&Vec>) -> Result<()> { + let bits = bits.ok_or_else(|| Error::InvalidBitfield(column.to_string()))?; + if bits.is_empty() { + return Err(Error::InvalidBitfield(column.to_string())); + } + let mut next_free = 0_i32; + for bit in bits { + // Groups must be ordered, non-overlapping and fit in 32 bits. + if bit.size <= 0 || bit.offset < next_free || bit.offset + bit.size > 32 { + return Err(Error::InvalidBitfield(column.to_string())); + } + next_free = bit.offset + bit.size; + } + Ok(()) +} diff --git a/rust/crates/odc/src/error.rs b/rust/crates/odc/src/error.rs new file mode 100644 index 00000000..f28865da --- /dev/null +++ b/rust/crates/odc/src/error.rs @@ -0,0 +1,72 @@ +//! Error types for odc operations. + +use odc_sys::ColumnType; + +/// Errors returned by odc operations. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// Typed odc C++ exception (auto-generated in `odc-sys`). + #[error(transparent)] + Odc(#[from] odc_sys::Error), + + /// Typed eckit C++ exception (e.g. from `DataHandle` operations). + #[error(transparent)] + Eckit(#[from] eckit::Error), + + /// Polars error while building or consuming a `DataFrame`. + #[error(transparent)] + Polars(#[from] polars::prelude::PolarsError), + + /// A `DataFrame` column has a dtype that cannot be encoded to ODB-2. + #[error("column '{column}' has unsupported dtype {dtype} for ODB encoding")] + UnsupportedDtype { column: String, dtype: String }, + + /// A requested ODB column cannot be decoded (e.g. type `Ignore`). + #[error("column '{column}' has unsupported ODB type {column_type:?}")] + UnsupportedColumnType { + column: String, + column_type: ColumnType, + }, + + /// A type override in [`WriteOptions::types`](crate::WriteOptions::types) + /// is not compatible with the column's dtype. + #[error("column '{column}': cannot encode {from:?} data as {to:?}")] + InvalidTypeOverride { + column: String, + from: ColumnType, + to: ColumnType, + }, + + /// Column not found in the frame. + #[error("column not found: {0}")] + ColumnNotFound(String), + + /// Encoding requires at least one row and one column. + #[error("cannot encode an empty DataFrame")] + EmptyDataFrame, + + /// Invalid bitfield specification (missing, oversized or overlapping). + #[error("invalid bitfield specification for column '{0}'")] + InvalidBitfield(String), + + /// I/O error. + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), +} + +/// Result type alias for odc operations. +pub type Result = std::result::Result; + +impl From for Error { + fn from(e: odc_sys::Exception) -> Self { + // The trycatch bridge prefixes messages with the exception's + // namespace — try odc's typed errors first, then eckit's. + if let Some(err) = odc_sys::Error::try_from_cxx(&e) { + return Self::Odc(err); + } + if let Some(err) = eckit::Error::try_from_cxx(&e) { + return Self::Eckit(err); + } + Self::Odc(odc_sys::Error::Other(e.what().to_string())) + } +} diff --git a/rust/crates/odc/src/frame.rs b/rust/crates/odc/src/frame.rs new file mode 100644 index 00000000..973656ef --- /dev/null +++ b/rust/crates/odc/src/frame.rs @@ -0,0 +1,125 @@ +//! [`Frame`] — a decodable chunk of an ODB-2 stream. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use polars::prelude::DataFrame; + +use crate::decode; +use crate::error::Result; +use crate::reader::ReaderShared; +use odc_sys::ColumnInfo; + +/// Options for decoding a [`Frame`] into a `DataFrame`. +#[derive(Debug, Clone)] +pub struct DecodeOptions { + /// Columns to decode, in the requested order. `None` decodes all. + pub columns: Option>, + /// Number of decode threads. Parallelism applies across the physical + /// frames of an aggregated logical frame. + pub threads: usize, +} + +impl Default for DecodeOptions { + fn default() -> Self { + Self { + columns: None, + threads: 1, + } + } +} + +/// A viewport onto a chunk of contiguous, compatible data within an ODB-2 +/// stream — possibly a logical frame aggregating several physical frames. +/// +/// Column metadata and properties are available without decoding; the data +/// itself decodes into a Polars `DataFrame` via [`Frame::dataframe`]. +pub struct Frame { + inner: odc_sys::UniquePtr, + columns: Vec, + properties: BTreeMap, + // Frames read lazily from the reader's stream — keep it alive. + _reader: Arc, +} + +impl Frame { + pub(crate) fn new( + inner: odc_sys::UniquePtr, + reader: Arc, + ) -> Result { + let columns = inner.column_info()?; + let properties = inner + .properties()? + .into_iter() + .map(|p| (p.key, p.value)) + .collect(); + Ok(Self { + inner, + columns, + properties, + _reader: reader, + }) + } + + pub(crate) fn wrapper(&self) -> &odc_sys::FrameWrapper { + &self.inner + } + + /// Number of rows. + #[must_use] + pub fn row_count(&self) -> usize { + self.inner.row_count() + } + + /// Number of columns. + #[must_use] + pub fn column_count(&self) -> usize { + self.inner.column_count() + } + + /// Column metadata, in frame order. + #[must_use] + pub fn columns(&self) -> &[ColumnInfo] { + &self.columns + } + + /// Metadata of the named column. + #[must_use] + pub fn column(&self, name: &str) -> Option<&ColumnInfo> { + self.columns.iter().find(|c| c.name == name) + } + + /// Whether the frame has a column with this name. + #[must_use] + pub fn has_column(&self, name: &str) -> bool { + self.columns.iter().any(|c| c.name == name) + } + + /// Key/value properties encoded in the frame. + #[must_use] + pub const fn properties(&self) -> &BTreeMap { + &self.properties + } + + /// Decode all columns into a `DataFrame`. + /// + /// Missing values become nulls; see [`crate::read_odb`] for the full + /// type mapping. + /// + /// # Errors + /// + /// Fails if the underlying stream cannot be read or decoded. + pub fn dataframe(&self) -> Result { + decode::dataframe(self, &DecodeOptions::default()) + } + + /// Decode selected columns into a `DataFrame`. + /// + /// # Errors + /// + /// Fails if a requested column does not exist or the underlying stream + /// cannot be read or decoded. + pub fn dataframe_with(&self, options: &DecodeOptions) -> Result { + decode::dataframe(self, options) + } +} diff --git a/rust/crates/odc/src/lib.rs b/rust/crates/odc/src/lib.rs new file mode 100644 index 00000000..d7f7523c --- /dev/null +++ b/rust/crates/odc/src/lib.rs @@ -0,0 +1,152 @@ +//! Safe Rust wrapper for ECMWF's odc (ODB-2 encoder/decoder) library. +//! +//! ODB-2 data decodes into [Polars](https://pola.rs) `DataFrame`s and +//! encodes from them, following the same model as +//! [pyodc](https://github.com/ecmwf/pyodc): +//! +//! ```no_run +//! // One DataFrame per logical frame, or concatenated: +//! let df = odc::read_odb_single("data.odb", &odc::ReadOptions::default())?; +//! println!("{df}"); +//! +//! odc::write_odb(&df, "copy.odb", &odc::WriteOptions::default())?; +//! # Ok::<(), odc::Error>(()) +//! ``` +//! +//! For streaming access use [`Reader`] and iterate [`Frame`]s, inspecting +//! column metadata and properties before deciding what to decode. +//! +//! # Type mapping +//! +//! | ODB type | decodes to | encoded from | +//! |------------|------------|------------------------------------------| +//! | `Integer` | `Int64` | `Int64` (smaller ints/`Boolean` widened) | +//! | `Double` | `Float64` | `Float64` | +//! | `Real` | `Float32` | `Float32` | +//! | `String` | `String` | `String` | +//! | `Bitfield` | `Int64` | `Int64` + [`WriteOptions::bitfields`] | +//! +//! ODB missing values map to nulls in both directions (bitfields excepted). +//! +//! # Process-global state +//! +//! odc stores its integer behaviour and missing-value sentinels globally. +//! This crate pins integers-as-longs on first use, so INTEGER and BITFIELD +//! columns decode as `i64`. Other in-process users of the odc C++ library +//! observe the same setting. + +mod decode; +mod encode; +mod error; +mod frame; +mod reader; + +pub use encode::{WriteOptions, write_odb, write_odb_to}; +pub use error::{Error, Result}; +pub use frame::{DecodeOptions, Frame}; +pub use odc_sys::{Bit, ColumnInfo, ColumnType, Property}; +pub use polars; +pub use reader::{Frames, Reader, ReaderOptions}; + +use std::path::Path; + +use polars::prelude::DataFrame; + +/// One-time process-global initialization, called by every public entry +/// point: eckit runtime (with the Rust log bridge) and integers-as-longs +/// decode behaviour. +pub(crate) fn init() { + static ONCE: std::sync::Once = std::sync::Once::new(); + ONCE.call_once(|| { + eckit::init(); + odc_sys::SettingsWrapper::treat_integers_as_doubles(false); + }); +} + +/// Release version of the odc C++ library, e.g. `1.6.3`. +#[must_use] +pub fn version() -> String { + init(); + odc_sys::SettingsWrapper::version() +} + +/// Version control checksum of the odc C++ library. +#[must_use] +pub fn vcs_version() -> String { + init(); + odc_sys::SettingsWrapper::gitsha1() +} + +/// The sentinel value marking a missing integer in ODB-2 data. +#[must_use] +pub fn integer_missing_value() -> i64 { + init(); + odc_sys::SettingsWrapper::integer_missing_value() +} + +/// The sentinel value marking a missing double in ODB-2 data. +#[must_use] +pub fn double_missing_value() -> f64 { + init(); + odc_sys::SettingsWrapper::double_missing_value() +} + +/// Options for [`read_odb`] and [`read_odb_single`]. +#[derive(Debug, Clone)] +pub struct ReadOptions { + /// Columns to decode, in the requested order. `None` decodes all. + pub columns: Option>, + /// Aggregate consecutive compatible physical frames into logical frames. + pub aggregated: bool, + /// Number of decode threads per frame. + pub threads: usize, +} + +impl Default for ReadOptions { + fn default() -> Self { + Self { + columns: None, + aggregated: true, + threads: 1, + } + } +} + +/// Decode an ODB-2 file into one `DataFrame` per logical frame. +/// +/// # Errors +/// +/// Fails if the file cannot be opened, is not valid ODB-2, or a requested +/// column does not exist. +pub fn read_odb(path: impl AsRef, options: &ReadOptions) -> Result> { + let reader_options = ReaderOptions { + aggregated: options.aggregated, + row_limit: None, + }; + let decode_options = DecodeOptions { + columns: options.columns.clone(), + threads: options.threads, + }; + Reader::from_path_with(path, &reader_options)? + .frames() + .map(|frame| frame?.dataframe_with(&decode_options)) + .collect() +} + +/// Decode an ODB-2 file into a single `DataFrame`, concatenating all frames. +/// +/// Returns an empty `DataFrame` for an empty source. +/// +/// # Errors +/// +/// Fails like [`read_odb`], or if frames have incompatible schemas. +pub fn read_odb_single(path: impl AsRef, options: &ReadOptions) -> Result { + let mut frames = read_odb(path, options)?.into_iter(); + let Some(mut df) = frames.next() else { + return Ok(DataFrame::default()); + }; + for frame in frames { + df.vstack_mut(&frame)?; + } + Ok(df) +} diff --git a/rust/crates/odc/src/reader.rs b/rust/crates/odc/src/reader.rs new file mode 100644 index 00000000..9485d7a9 --- /dev/null +++ b/rust/crates/odc/src/reader.rs @@ -0,0 +1,158 @@ +//! Reading ODB-2 data: [`Reader`] and its frame iterator. + +use std::path::Path; +use std::sync::Arc; + +use parking_lot::Mutex; + +use crate::error::{Error, Result}; +use crate::frame::Frame; +use crate::init; + +/// Options for opening a [`Reader`]. +#[derive(Debug, Clone)] +pub struct ReaderOptions { + /// Aggregate consecutive compatible physical frames into logical frames. + pub aggregated: bool, + /// Maximum number of rows to aggregate into one logical frame. + pub row_limit: Option, +} + +impl Default for ReaderOptions { + fn default() -> Self { + Self { + aggregated: true, + row_limit: None, + } + } +} + +/// Shared reader state. Frames decode lazily from the reader's stream, so +/// every [`Frame`] holds an `Arc` of this to keep the stream (and the source +/// handle) alive. +pub struct ReaderShared { + // Field order matters: the C++ reader borrows the source handle, so + // `inner` must drop before `_source`. + pub(crate) inner: Mutex>, + _source: Option>>, +} + +/// Owns an ODB-2 data stream and yields its [`Frame`]s. +/// +/// # Thread safety +/// +/// `Reader` is `Send + Sync`; stream access is serialized through a mutex. +/// Frames may be decoded from other threads while the reader advances. +pub struct Reader { + shared: Arc, +} + +impl Reader { + /// Open an ODB-2 file with default options. + /// + /// # Errors + /// + /// Fails if the file cannot be opened or is not valid ODB-2. + pub fn from_path(path: impl AsRef) -> Result { + Self::from_path_with(path, &ReaderOptions::default()) + } + + /// Open an ODB-2 file. + /// + /// # Errors + /// + /// Fails if the file cannot be opened or is not valid ODB-2. + pub fn from_path_with(path: impl AsRef, options: &ReaderOptions) -> Result { + init(); + let path = path.as_ref().to_str().ok_or_else(|| { + Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "path is not valid UTF-8", + )) + })?; + let inner = odc_sys::ReaderWrapper::from_path( + path, + options.aggregated, + options.row_limit.unwrap_or(-1), + )?; + Ok(Self { + shared: Arc::new(ReaderShared { + inner: Mutex::new(inner), + _source: None, + }), + }) + } + + /// Read ODB-2 data from an eckit [`DataHandle`](eckit::DataHandle) + /// (file, buffer, multi-file, byte range, …). + /// + /// The handle must not be open — the reader opens it for reading and it + /// stays owned by the reader for its whole lifetime. + /// + /// # Errors + /// + /// Fails if the handle cannot be opened or is not valid ODB-2. + pub fn from_handle( + mut handle: eckit::DataHandle, + options: &ReaderOptions, + ) -> Result { + init(); + let inner = odc_sys::ReaderWrapper::from_handle( + handle.inner_mut()?, + options.aggregated, + options.row_limit.unwrap_or(-1), + )?; + Ok(Self { + shared: Arc::new(ReaderShared { + inner: Mutex::new(inner), + _source: Some(Mutex::new(handle)), + }), + }) + } + + /// Iterator over the frames of the stream. + /// + /// The iterator advances the underlying stream: each frame is yielded + /// once, and a second `frames()` call continues where the first stopped. + #[must_use] + pub fn frames(&self) -> Frames { + Frames { + shared: Arc::clone(&self.shared), + exhausted: false, + } + } +} + +/// Iterator over the [`Frame`]s of a [`Reader`]. +pub struct Frames { + shared: Arc, + exhausted: bool, +} + +impl Iterator for Frames { + type Item = Result; + + fn next(&mut self) -> Option { + if self.exhausted { + return None; + } + let next = self.shared.inner.lock().pin_mut().next_frame(); + match next { + Err(e) => { + self.exhausted = true; + Some(Err(e.into())) + } + Ok(ptr) if ptr.is_null() => { + self.exhausted = true; + None + } + Ok(ptr) => match Frame::new(ptr, Arc::clone(&self.shared)) { + Ok(frame) => Some(Ok(frame)), + Err(e) => { + self.exhausted = true; + Some(Err(e)) + } + }, + } + } +}