datafusion_opensearch/
lib.rs

1//! A DataFusion [`TableProvider`] over an OpenSearch (or Elasticsearch) index.
2//!
3//! `SELECT … FROM <index> WHERE …` planned by DataFusion becomes an OpenSearch `_search`: the
4//! WHERE is pushed into the query DSL ([`pushdown::expr_to_query`]), the projection into
5//! `_source`, and the LIMIT into the page size. The result is **streamed page by page** through a
6//! scroll cursor ([`OpenSearchExec`]), so a scan is complete at any index size with memory bounded
7//! by the page size. Filters that cannot be translated exactly are reported
8//! [`TableProviderFilterPushDown::Unsupported`], so DataFusion re-applies them in-memory — partial
9//! pushdown is therefore always correct and never widens the result set.
10//!
11//! Three ways in:
12//!
13//! - SQL only: register [`OpenSearchTableProviderFactory`] and
14//!   `CREATE EXTERNAL TABLE docs STORED AS OPENSEARCH LOCATION 'http://localhost:9200/my-index'`.
15//! - A whole cluster: [`OpenSearchSchemaProvider`] lists every index as a table (`SHOW TABLES`).
16//! - One index: [`OpenSearchTableFactory::table_provider`] (schema from the `_mapping`) or
17//!   [`OpenSearchTableProvider::new`] with a declared Arrow schema.
18//!
19//! ```no_run
20//! use datafusion::prelude::SessionContext;
21//! use datafusion_opensearch::OpenSearchTableProviderFactory;
22//!
23//! # async fn run() -> datafusion::error::Result<()> {
24//! let ctx = SessionContext::new();
25//! OpenSearchTableProviderFactory::new().register(&ctx);
26//! ctx.sql("CREATE EXTERNAL TABLE docs STORED AS OPENSEARCH LOCATION 'http://localhost:9200/my-index'")
27//!     .await?
28//!     .collect()
29//!     .await?;
30//! ctx.sql("SELECT id, status FROM docs WHERE speed > 40 AND status = 'OK' LIMIT 100").await?.show().await?;
31//! # Ok(())
32//! # }
33//! ```
34//!
35//! With the `udf` feature, OpenSearch-native predicates — full-text `match` and the geo bounding
36//! box / distance / polygon queries — are available as scalar UDFs ([`udf`]) that the provider
37//! pushes down as native queries.
38//!
39//! [`TableProvider`]: datafusion::catalog::TableProvider
40//! [`TableProviderFilterPushDown::Unsupported`]: datafusion::logical_expr::TableProviderFilterPushDown::Unsupported
41
42/// Every Rust code block in the README is compiled as a doctest (`no_run`: they need a server).
43#[cfg(doctest)]
44#[doc = include_str!("../README.md")]
45pub struct ReadmeDoctests;
46
47pub mod catalog;
48pub mod client;
49pub mod exec;
50pub mod pushdown;
51pub mod schema;
52pub mod table;
53#[cfg(feature = "udf")]
54pub mod udf;
55
56pub use catalog::OpenSearchSchemaProvider;
57pub use client::{OpenSearchClient, SearchPage};
58pub use exec::OpenSearchExec;
59pub use schema::schema_from_mapping;
60pub use table::{
61    OpenSearchTableFactory, OpenSearchTableProvider, OpenSearchTableProviderFactory, DEFAULT_PAGE_SIZE, FILE_TYPE,
62};
63#[cfg(feature = "udf")]
64pub use udf::{
65    os_geo_bbox_udf, os_geo_distance_udf, os_geo_polygon_udf, os_match_udf, OS_GEO_BBOX, OS_GEO_DISTANCE,
66    OS_GEO_POLYGON, OS_MATCH,
67};
68
69use datafusion::error::DataFusionError;
70use snafu::Snafu;
71
72/// Errors raised while talking to OpenSearch or shaping its responses. Inside a DataFusion plan
73/// they surface as [`DataFusionError::External`].
74#[derive(Debug, Snafu)]
75#[snafu(visibility(pub(crate)))]
76pub enum Error {
77    #[snafu(display("request to {url} failed: {source}"))]
78    Http { url: String, source: reqwest::Error },
79
80    #[snafu(display("OpenSearch returned {status} for {url}: {body}"))]
81    Status { status: u16, url: String, body: String },
82
83    #[snafu(display("index '{index}' does not exist"))]
84    IndexNotFound { index: String },
85
86    #[snafu(display("could not decode the OpenSearch response from {url}: {source}"))]
87    Decode { url: String, source: reqwest::Error },
88
89    #[snafu(display("the _mapping response for '{index}' has no 'mappings.properties'"))]
90    MalformedMapping { index: String },
91}
92
93/// Convenience alias used throughout the crate.
94pub type Result<T, E = Error> = std::result::Result<T, E>;
95
96impl From<Error> for DataFusionError {
97    fn from(e: Error) -> Self {
98        DataFusionError::External(Box::new(e))
99    }
100}
datafusion_opensearch/lib.rs

datafusion_opensearch/
lib.rs
