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 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/config.toml b/rust/.cargo/config.toml new file mode 100644 index 00000000..b4069aba --- /dev/null +++ b/rust/.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/rust/Cargo.toml b/rust/Cargo.toml new file mode 100644 index 00000000..5221023f --- /dev/null +++ b/rust/Cargo.toml @@ -0,0 +1,33 @@ +[workspace] +resolver = "2" +members = ["crates/odc", "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 = { 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 +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" +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/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..a282b23a --- /dev/null +++ b/rust/crates/odc-sys/build.rs @@ -0,0 +1,203 @@ +use std::env; +use std::path::{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/OdcBridge.h"); + 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"); + + 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(); + } +} + +/// 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_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"); + + cxx_build::bridge("src/lib.rs") + .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) + .include(crate_dir.join("cpp")) + .include(&out_dir) // for odc_exceptions.h (generated) + .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_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); + + generate_exceptions(&include_dir); + + cxx_build::bridge("src/lib.rs") + .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) + .include(crate_dir.join("cpp")) + .include(&out_dir) // for odc_exceptions.h (generated) + .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/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..480e2bb1 --- /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(Bit{rust::String(bit.name), bit.size, bit.offset}); + } + 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; + for (const auto& [key, value] : frame_.properties()) { + result.push_back(Property{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..bf8bc159 --- /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 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 +/// 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.h b/rust/crates/odc-sys/cpp/OdcBridge.h new file mode 100644 index 00000000..0338d14b --- /dev/null +++ b/rust/crates/odc-sys/cpp/OdcBridge.h @@ -0,0 +1,17 @@ +// 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 + +// 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 new file mode 100644 index 00000000..1e483055 --- /dev/null +++ b/rust/crates/odc-sys/src/lib.rs @@ -0,0 +1,231 @@ +//! 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; + +// 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"), + ("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 { + /// 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, + } + + /// A bit group within a bitfield column — C++ `odc::api::ColumnInfo::Bit`. + #[derive(Debug, Clone, PartialEq, Eq)] + struct Bit { + name: String, + /// Bit group size in bits. + size: i32, + /// Bit group offset in bits. + offset: i32, + } + + /// Metadata for one column of a frame — C++ `odc::api::ColumnInfo`. + #[derive(Debug, Clone)] + 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, + } + + /// A key/value property encoded in a frame. + #[derive(Debug, Clone, PartialEq, Eq)] + struct Property { + key: String, + value: String, + } + + 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 + #[namespace = "odc::api"] + type ColumnType; + + // Cross-crate ExternType from eckit-sys + #[namespace = "eckit_bridge"] + type DataHandleWrapper = eckit_sys::DataHandleWrapper; + + // ==================== ReaderWrapper ==================== + + type ReaderWrapper; + + /// Open an ODB-2 file for reading. + #[Self = "ReaderWrapper"] + fn from_path( + path: &str, + aggregated: bool, + rowlimit: i64, + ) -> 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>; + + /// Next frame in the stream; null when exhausted. + fn next_frame(self: Pin<&mut ReaderWrapper>) -> Result>; + + // ==================== FrameWrapper ==================== + + 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>; + + // ==================== DecoderWrapper ==================== + + type DecoderWrapper; + + #[Self = "DecoderWrapper"] + #[must_use] + 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, + 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"] + #[must_use] + 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 set_property(self: Pin<&mut EncoderWrapper>, key: &str, value: &str); + + /// 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<()>; + + // ==================== 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"] + #[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; + } +} + +pub use cxx::{Exception, UniquePtr}; +pub use ffi::*; + +// 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::{DecoderWrapper, EncoderWrapper, FrameWrapper, ReaderWrapper}; + unsafe impl Send for ReaderWrapper {} + unsafe impl Send for FrameWrapper {} + unsafe impl Send for DecoderWrapper {} + unsafe impl Send for EncoderWrapper {} +} 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)) + } + }, + } + } +}