feat: merge schemas as numpy promotes types

CHANGELOG.md

@@ -23,6 +23,30 @@ tag. Releases before 2.0.0 are recorded in the
   where it used to be a `404`. The page carries the colors, the type and the card layout of the
   documentation site, and it loads nothing from the network, so it also renders on a server with
   no route out. Swagger keeps its own path, so a bookmark to `/swagger` is unaffected.
+- **`BEACON_TYPE_WIDENING_STRATEGY=numpy` merges schemas as numpy promotes types.** The schema
+  merge widens a column inside one family: a wider integer, a finer timestamp, a longer string. It
+  refuses every other pair, and it reads every integer beside a `Float32` as `Float64`, because a
+  lattice holds no other answer that is free of the listing order. A collection written by numpy or
+  xarray expects the rules of `numpy.result_type` instead. The new strategy applies them: a boolean
+  joins the numbers (`bool` + `int8` is `int8`), `Float16` joins the floats, a narrow integer beside
+  a `Float32` stays a `Float32`, a number or a boolean beside a string reads as text, and a date
+  beside a timestamp is a timestamp at the finer unit. numpy resolves a set of types at once, and
+  the answer differs from a chain of pairs: `int8` + `uint8` is `int16` and `int16` + `float16` is
+  `float32`, yet `result_type(int8, uint8, float16)` is `float16`. The strategy therefore gathers
+  the types of each column across every file and resolves the set once, so the listing order does
+  not change the result. The cost is one pass over every schema, as `keep_first` pays. Four numpy
+  rules stay behind, because Arrow has no cast for them: an integer beside a `timedelta64`, a
+  `timedelta64` beside a `datetime64`, and a number or a text string beside a byte string are
+  conflicts, and a time of day keeps the default chain. One limit is the CSV reader's: a text file
+  holds no types, so it parses each column as the merged type. A number beside a string reads as
+  the text of the file, and a boolean literal beside a number fails at read time, where every typed
+  format casts `true` to `1`. `BEACON_TYPE_WIDENING_ON_CONFLICT` applies under either strategy.
+  `default` keeps the merge every server ran before, and an unknown value logs a warning and reads
+  as `default`. The strategy is an `ArrowTypeWideningStrategy` like the default one, so an embedded
+  build passes `NumpyArrowTypeWidening` to `RuntimeBuilder::with_type_widening` or
+  `OpenOptions::with_type_widening`. See
+  [Configuration](docs/docs/2.0.0-rc5/server/configuration.md#query-engine) and
+  [Troubleshooting](docs/docs/2.0.0-rc5/troubleshooting.md#a-column-has-two-types-across-the-files).
 - **`BEACON_TYPE_WIDENING_ON_CONFLICT` settles a column that no type holds.** A collection can
   type one column as a number in one file and as a string in another. No type holds both, so the
   schema merge refused the whole table and the table answered no query: `Incompatible types for

beacon-db/beacon-core/src/embedded.rs

@@ -35,7 +35,7 @@ use std::sync::Arc;
 use beacon_auth::{AuthIdentity, Credential, ANONYMOUS_USERNAME};
 use beacon_common::FileStatsConfig;
 use beacon_datafusion_ext::listing_factory::DefaultStore;
-use beacon_datafusion_ext::type_widening::TypeConflict;
+use beacon_datafusion_ext::type_widening::{ArrowTypeWideningStrategy, TypeConflict};
 use datafusion::scalar::ScalarValue;
 use tokio::runtime::Handle;
 
@@ -234,6 +234,11 @@ pub struct OpenOptions {
     /// The container file is still opened with an exclusive lock, so this is a per-connection
     /// writability guarantee, not (yet) multi-process concurrent access.
     pub read_only: bool,
+    /// A merge rule of the embedder's own, such as
+    /// [`NumpyArrowTypeWidening`](beacon_datafusion_ext::type_widening::NumpyArrowTypeWidening).
+    /// It replaces the default rule and [`Self::type_conflict`]. `None` takes the
+    /// default rule with that setting.
+    pub type_widening: Option<Arc<dyn ArrowTypeWideningStrategy>>,
     /// What a schema merge does with a column that no type holds, e.g. a number
     /// in one file and a string in another. Defaults to [`TypeConflict::Fail`],
     /// which refuses such a collection.
@@ -287,6 +292,15 @@ impl OpenOptions {
         self
     }
 
+    /// Set the rule for every schema merge, in place of the default rule.
+    ///
+    /// The rule carries its own conflict setting, so
+    /// [`with_type_conflict`](Self::with_type_conflict) does not apply to it.
+    pub fn with_type_widening(mut self, strategy: Arc<dyn ArrowTypeWideningStrategy>) -> Self {
+        self.type_widening = Some(strategy);
+        self
+    }
+
     /// Set what a schema merge does with a column that no type holds.
     ///
     /// [`TypeConflict::KeepFirst`] reads the type of the first file, and null
@@ -392,7 +406,10 @@ impl Database {
         if options.nd_pipeline {
             builder = builder.with_nd_pipeline();
         }
-        builder = builder.with_type_conflict(options.type_conflict);
+        builder = match options.type_widening.clone() {
+            Some(strategy) => builder.with_type_widening(strategy),
+            None => builder.with_type_conflict(options.type_conflict),
+        };
         if let Some(datasets) = &options.datasets {
             builder = builder.with_default_store(datasets.url.clone(), datasets.root.clone());
         }

beacon-db/beacon-core/src/runtime_builder.rs

@@ -255,6 +255,8 @@ impl RuntimeBuilder {
     /// The default is
     /// [`DefaultArrowTypeWidening`](beacon_datafusion_ext::type_widening::DefaultArrowTypeWidening).
     /// It unions the fields that agree and refuses the rest.
+    /// [`NumpyArrowTypeWidening`](beacon_datafusion_ext::type_widening::NumpyArrowTypeWidening)
+    /// promotes as `numpy.result_type` does.
     pub fn with_type_widening(mut self, strategy: Arc<dyn ArrowTypeWideningStrategy>) -> Self {
         self.type_widening = Some(strategy);
         self

beacon-db/beacon-datafusion-ext/src/type_widening.rs

@@ -108,6 +108,19 @@
 //! [`Fail`]: TypeConflict::Fail
 //! [`KeepFirst`]: TypeConflict::KeepFirst
 //!
+//! # Two strategies
+//!
+//! [`DefaultArrowTypeWidening`] applies the rules above. [`NumpyArrowTypeWidening`]
+//! promotes as `numpy.result_type` does: a boolean joins the numbers, `Float16`
+//! joins the floats, a narrow integer beside a `Float32` stays a `Float32`, and a
+//! number beside a string reads as text. The [`numpy`](self::numpy) module holds
+//! its rule table and the pairs where it leaves numpy. The setting for a conflict
+//! applies under either strategy.
+//!
+//! A deployment picks one through [`RuntimeBuilder::with_type_widening`]. The
+//! server builds it from `BEACON_TYPE_WIDENING_STRATEGY`, which names `default`
+//! or `numpy`.
+//!
 //! # The source of a conflict
 //!
 //! A merge that refuses a column names the source of each type. A table over
@@ -159,6 +172,10 @@ use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema, SchemaRef, Tim
 use datafusion::catalog::Session;
 use object_store::ObjectMeta;
 
+pub mod numpy;
+
+pub use numpy::NumpyArrowTypeWidening;
+
 /// Below this count, one fold costs less than a split. A merge walks field
 /// names, so a thread must earn its start cost.
 const SEQUENTIAL_MERGE_LIMIT: usize = 64;
@@ -274,7 +291,9 @@ pub fn session_widening(session: &dyn Session) -> Arc<ArrowTypeWidening> {
         .unwrap_or_else(ArrowTypeWidening::default_extension)
 }
 
-pub trait ArrowTypeWideningStrategy: Send + Sync {
+/// The rule a merge applies. `Debug` lets a configuration that holds the rule
+/// print it.
+pub trait ArrowTypeWideningStrategy: std::fmt::Debug + Send + Sync {
     /// Merge these schemas into one, in the order given.
     ///
     /// Each schema names its source. Report both names for a column that two
@@ -290,7 +309,9 @@ pub trait ArrowTypeWideningStrategy: Send + Sync {
     /// [`DefaultArrowTypeWidening`] qualifies.
     ///
     /// Answer `false` for a rule that reads the order. One example keeps the first
-    /// type of a column. Such a merge gets one fold over every schema.
+    /// type of a column. Another resolves the set of types of a column at once,
+    /// which a chunk result would hide: see [`NumpyArrowTypeWidening`]. Such a
+    /// merge gets one fold over every schema.
     fn is_order_independent(&self) -> bool {
         true
     }
@@ -616,18 +637,28 @@ fn timestamp_super_type(
     right_unit: &TimeUnit,
     right_zone: &Option<Arc<str>>,
 ) -> Option<DataType> {
-    let zone = match (left_zone, right_zone) {
-        (None, None) => None,
-        (Some(zone), None) | (None, Some(zone)) => Some(Arc::clone(zone)),
-        (Some(left), Some(right)) if left == right => Some(Arc::clone(left)),
-        (Some(_), Some(_)) => Some(UTC.into()),
-    };
     let finer = if time_unit_rank(left_unit) >= time_unit_rank(right_unit) {
         left_unit
     } else {
         right_unit
     };
-    Some(DataType::Timestamp(finer.clone(), zone))
+    Some(DataType::Timestamp(
+        *finer,
+        zone_join(left_zone, right_zone),
+    ))
+}
+
+/// The zone that two timestamp columns read as. See [`timestamp_super_type`].
+///
+/// The rule is a lattice join: no zone sits below every zone, and [`UTC`] sits
+/// above every zone. The numpy strategy folds it over a set.
+fn zone_join(left: &Option<Arc<str>>, right: &Option<Arc<str>>) -> Option<Arc<str>> {
+    match (left, right) {
+        (None, None) => None,
+        (Some(zone), None) | (None, Some(zone)) => Some(Arc::clone(zone)),
+        (Some(left), Some(right)) if left == right => Some(Arc::clone(left)),
+        (Some(_), Some(_)) => Some(UTC.into()),
+    }
 }
 
 /// The zone that two other zones read as. `"+00:00"` names the same zone, and it
@@ -1821,6 +1852,7 @@ mod tests {
     fn an_order_sensitive_strategy_sees_every_schema() {
         use std::sync::atomic::{AtomicUsize, Ordering};
 
+        #[derive(Debug)]
         struct CountingStrategy {
             calls: AtomicUsize,
             seen: AtomicUsize,