From 59b9d226af75866058f237e653a6bcdf8f4aafde Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Thu, 2 Jul 2026 05:06:00 -0700 Subject: [PATCH 1/2] [Analytics Engine] Error on BIGINT arithmetic overflow instead of wrapping BIGINT (Int64) +, -, * overflow on the DataFusion backend silently wrapped (arrow's wrapping kernels) and returned HTTP 200 with a wrong value. Rewrite every Int64-operand +/-/* logical BinaryExpr into an overflow-checked scalar UDF (checked_{add,sub,mul}_i64) that calls arrow's checked kernels and errors on overflow, surfaced to the client as HTTP 400. - checked_arith UDFs (Immutable, return Int64) re-wrap the arrow overflow into a stable "BIGINT arithmetic overflow" DataFusionError. - checked_arith_rewrite walks the logical plan (map_expressions + Expr transform_up), gated to Int64-both-operands so float/decimal/uint/narrow-int arithmetic is untouched; NamePreserver keeps aggregate measure names stable. - Applied on every executor's executed plan and the indexed-path filter extraction plan (so where-clause arithmetic feeding parquet pushdown is checked too); nested inside comparisons only, so delegation is unaffected. - NativeErrorConverter maps the overflow to a 400 OpenSearchStatusException; AnalyticsTransportErrors tags it INVALID_ARGUMENT across Flight so the 400 survives the distributed-aggregate reduce instead of degrading to a 500. Calcite/v2 long-overflow parity is tracked separately. Signed-off-by: Kai Huang --- .../rust/src/checked_arith_rewrite.rs | 266 ++++++++++++++++++ .../rust/src/indexed_executor.rs | 9 + .../rust/src/lib.rs | 1 + .../rust/src/local_executor.rs | 4 + .../rust/src/query_executor.rs | 4 + .../rust/src/udf/checked_arith.rs | 246 ++++++++++++++++ .../rust/src/udf/mod.rs | 2 + .../be/datafusion/NativeErrorConverter.java | 26 +- .../datafusion/NativeErrorConverterTests.java | 30 ++ .../exec/AnalyticsTransportErrors.java | 29 +- .../exec/AnalyticsTransportErrorsTests.java | 62 ++++ .../analytics/qa/BigintOverflowIT.java | 157 +++++++++++ 12 files changed, 834 insertions(+), 2 deletions(-) create mode 100644 sandbox/plugins/analytics-backend-datafusion/rust/src/checked_arith_rewrite.rs create mode 100644 sandbox/plugins/analytics-backend-datafusion/rust/src/udf/checked_arith.rs create mode 100644 sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/BigintOverflowIT.java diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/checked_arith_rewrite.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/checked_arith_rewrite.rs new file mode 100644 index 0000000000000..1de709741b10b --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/checked_arith_rewrite.rs @@ -0,0 +1,266 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +//! Logical-plan rewrite that makes BIGINT (`Int64`) `+`, `-`, `*` overflow *error* instead of +//! silently wrapping on the DataFusion backend. +//! +//! DataFusion lowers `Operator::Plus/Minus/Multiply` to arrow's wrapping integer kernels, so +//! `9223372036854775807 * 2` produces a wrapped negative value with no error. PPL callers expect +//! an error. This pass rewrites every logical `BinaryExpr { op: Plus | Minus | Multiply }` whose +//! **both operands resolve to `Int64`** into a call to the overflow-checked UDF equivalent +//! (`checked_add_i64` / `checked_sub_i64` / `checked_mul_i64` — see [`crate::udf::checked_arith`]). +//! +//! # Why logical, not physical +//! +//! Walking the *logical* plan reaches arithmetic in every node uniformly — Projection exprs, +//! Filter predicates, Aggregate group/argument exprs (`sum(a*b)`), Window args, Sort keys, Join +//! conditions — because [`LogicalPlan::map_expressions`] visits each node's top-level exprs and +//! `Expr::transform_up` recurses into every sub-expr. A physical-plan pass would have to reach the +//! exprs through per-node accessors, which do not uniformly surface aggregate/window argument +//! expressions. +//! +//! # Type gate +//! +//! Only `Int64 op Int64` is rewritten. `Float32/Float64` (which wrap to ±INF by PPL parity), +//! `Decimal*`, `UInt64`, and narrower integers all pass through untouched. After the SQL-plugin +//! widening (sql#5603), the only integer arithmetic that reaches DataFusion and can still overflow +//! is `Int64 op Int64`, so this gate is both correct and sufficient. Note the analytics scan +//! boundary already narrows `UInt64 -> Int64` (see `schema_coerce::coerce_inferred_schema`), so +//! OpenSearch `long` columns resolve to `Int64` here. + +use std::sync::Arc; + +use datafusion::arrow::datatypes::DataType; +use datafusion::common::tree_node::{Transformed, TreeNode}; +use datafusion::common::{DFSchema, DFSchemaRef, Result}; +use datafusion::logical_expr::expr::ScalarFunction; +use datafusion::logical_expr::expr_rewriter::NamePreserver; +use datafusion::logical_expr::{BinaryExpr, Expr, ExprSchemable, LogicalPlan, Operator, ScalarUDF}; + +use crate::udf::checked_arith::{checked_add_udf, checked_mul_udf, checked_sub_udf}; + +/// Rewrite every `Int64 {+,-,*} Int64` `BinaryExpr` in `plan` into an overflow-checked UDF call. +/// Non-`Int64` arithmetic and non-`+/-/*` operators are left unchanged. +pub fn rewrite_checked_int64_arith(plan: LogicalPlan) -> Result { + plan.transform_up(|node| { + // Operands of a node's expressions are evaluated over the node's *input* columns, so + // resolve types against the merged input schema. For leaf nodes (no inputs) fall back to + // the node's own schema. + let schema = merged_input_schema(&node); + // Preserve each expression's output name: rewriting `sum(a * b)` to + // `sum(checked_mul_i64(a, b))` would otherwise let a later `recompute_schema` rename the + // output field, breaking any parent node that references it by the original derived name. + // NamePreserver re-aliases only when the name actually changed, and is a no-op for nodes + // whose expressions don't contribute to the output schema (Filter/Join/…). + let name_preserver = NamePreserver::new(&node); + node.map_expressions(|expr| { + let saved_name = name_preserver.save(&expr); + expr.transform_up(|e| rewrite_expr(e, &schema)) + .map(|t| t.update_data(|rewritten| saved_name.restore(rewritten))) + }) + }) + .map(|t| t.data) +} + +/// Merge the schemas of a node's inputs so operand `Column`s resolve. Leaf nodes have no inputs, +/// so their own schema is used (their exprs, if any, are typed over it). +fn merged_input_schema(node: &LogicalPlan) -> DFSchemaRef { + let inputs = node.inputs(); + if inputs.is_empty() { + return Arc::new(node.schema().as_ref().clone()); + } + let mut merged = DFSchema::empty(); + for child in inputs { + // merge() is order-preserving and dedups by qualified name; ignore duplicate-column errors. + merged.merge(child.schema()); + } + Arc::new(merged) +} + +fn rewrite_expr(expr: Expr, schema: &DFSchema) -> Result> { + let Expr::BinaryExpr(BinaryExpr { left, op, right }) = &expr else { + return Ok(Transformed::no(expr)); + }; + if !matches!(op, Operator::Plus | Operator::Minus | Operator::Multiply) { + return Ok(Transformed::no(expr)); + } + // Resolve operand types. If either operand's type cannot be resolved against this schema + // (an edge case such as a correlated column not present in the merged input schema), leave + // the expression unchanged rather than failing the whole query — a missed rewrite falls back + // to today's wrapping behavior, whereas propagating the error would break a working query. + let (Ok(lt), Ok(rt)) = (left.get_type(schema), right.get_type(schema)) else { + return Ok(Transformed::no(expr)); + }; + if lt != DataType::Int64 || rt != DataType::Int64 { + return Ok(Transformed::no(expr)); + } + + // Consume `expr` to move the boxed operands into the UDF call without cloning. + let Expr::BinaryExpr(BinaryExpr { left, op, right }) = expr else { + unreachable!("matched BinaryExpr above"); + }; + let udf: Arc = match op { + Operator::Plus => checked_add_udf(), + Operator::Minus => checked_sub_udf(), + Operator::Multiply => checked_mul_udf(), + _ => unreachable!("op guarded above"), + }; + let call = Expr::ScalarFunction(ScalarFunction::new_udf(udf, vec![*left, *right])); + Ok(Transformed::yes(call)) +} + +#[cfg(test)] +mod tests { + use super::*; + use datafusion::arrow::datatypes::Field; + use datafusion::logical_expr::{col, lit, LogicalPlanBuilder}; + + /// Count how many `ScalarFunction`s named `name` appear anywhere in the plan's expressions. + fn count_udf(plan: &LogicalPlan, name: &str) -> usize { + use std::cell::Cell; + let count = Cell::new(0usize); + plan.apply(|node| { + node.apply_expressions(|expr| { + expr.apply(|e| { + if let Expr::ScalarFunction(sf) = e { + if sf.func.name() == name { + count.set(count.get() + 1); + } + } + Ok(datafusion::common::tree_node::TreeNodeRecursion::Continue) + }) + }) + }) + .unwrap(); + count.get() + } + + /// A schema-carrying `EmptyRelation` builder; enough to type-resolve column operands for the + /// rewrite without a real table provider. + fn scan(fields: Vec) -> LogicalPlanBuilder { + use datafusion::logical_expr::{EmptyRelation, LogicalPlan}; + let arrow_schema = datafusion::arrow::datatypes::Schema::new(fields); + let schema = Arc::new(DFSchema::try_from(arrow_schema).unwrap()); + LogicalPlanBuilder::new(LogicalPlan::EmptyRelation(EmptyRelation { + produce_one_row: false, + schema, + })) + } + + fn int_float_scan() -> LogicalPlanBuilder { + scan(vec![ + Field::new("a", DataType::Int64, true), + Field::new("b", DataType::Int64, true), + Field::new("c", DataType::Int64, true), + Field::new("f", DataType::Float64, true), + Field::new("g", DataType::Float64, true), + ]) + } + + #[test] + fn rewrites_int64_mul_in_projection() { + let plan = int_float_scan() + .project(vec![(col("a") * col("b")).alias("x")]) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + assert_eq!(count_udf(&out, "checked_mul_i64"), 1); + assert_eq!(count_udf(&out, "checked_add_i64"), 0); + } + + #[test] + fn rewrites_int64_add_and_sub() { + let plan = int_float_scan() + .project(vec![ + (col("a") + col("b")).alias("s"), + (col("a") - col("b")).alias("d"), + ]) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + assert_eq!(count_udf(&out, "checked_add_i64"), 1); + assert_eq!(count_udf(&out, "checked_sub_i64"), 1); + } + + #[test] + fn leaves_float_mul_untouched() { + let plan = int_float_scan() + .project(vec![(col("f") * col("g")).alias("x")]) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + assert_eq!(count_udf(&out, "checked_mul_i64"), 0); + } + + #[test] + fn leaves_divide_untouched() { + let plan = int_float_scan() + .project(vec![(col("a") / col("b")).alias("x")]) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + // No checked_* udf: DIVIDE is not in {Plus,Minus,Multiply}. + assert_eq!(count_udf(&out, "checked_mul_i64"), 0); + assert_eq!(count_udf(&out, "checked_add_i64"), 0); + assert_eq!(count_udf(&out, "checked_sub_i64"), 0); + } + + #[test] + fn rewrites_arithmetic_in_filter_predicate() { + // Proves node-agnostic coverage: the arithmetic lives in a Filter, not a Projection. + let plan = int_float_scan() + .filter((col("a") * col("b")).gt(lit(5i64))) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + assert_eq!(count_udf(&out, "checked_mul_i64"), 1); + } + + #[test] + fn rewrites_nested_arithmetic_bottom_up() { + // (a * b) + c -> checked_add(checked_mul(a, b), c) + let plan = int_float_scan() + .project(vec![((col("a") * col("b")) + col("c")).alias("x")]) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + assert_eq!(count_udf(&out, "checked_mul_i64"), 1); + assert_eq!(count_udf(&out, "checked_add_i64"), 1); + } + + #[test] + fn rewrites_arithmetic_in_sort_key() { + // Arithmetic inside a Sort key must be reached (node-agnostic coverage beyond + // Projection/Filter — the same map_expressions traversal reaches Aggregate/Window args). + let plan = int_float_scan() + .sort(vec![(col("a") * col("b")).sort(true, false)]) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + assert_eq!(count_udf(&out, "checked_mul_i64"), 1); + } + + #[test] + fn leaves_int64_literal_times_float_untouched() { + // Mixed Int64 * Float64 -> operands are (Int64, Float64); gate requires BOTH Int64. + let plan = int_float_scan() + .project(vec![(col("a") * col("f")).alias("x")]) + .unwrap() + .build() + .unwrap(); + let out = rewrite_checked_int64_arith(plan).unwrap(); + assert_eq!(count_udf(&out, "checked_mul_i64"), 0); + } +} diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs index b3484fbe2c2b9..9067aebe1dc5a 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs @@ -960,6 +960,13 @@ async unsafe fn execute_indexed_with_context_inner( let plan = Plan::decode(substrait_bytes.as_slice()) .map_err(|e| DataFusionError::Execution(format!("decode substrait: {}", e)))?; let logical_plan = from_substrait_plan(&ctx.state(), &plan).await?; + // Make BIGINT +,-,* overflow error instead of wrapping in this plan too, since its filter + // predicates seed the parquet pushdown predicate (`where long1 * long2 > k` would otherwise be + // scan-filtered with wrapping arithmetic, dropping overflowing rows silently before the executed + // plan's checked filter runs). Checked arithmetic appears only nested inside a comparison operand + // — never as a top-level AND/OR or COLLECTOR/DELEGATION_POSSIBLE marker — so delegation + // classification (expr_to_bool_tree) is unaffected. + let logical_plan = crate::checked_arith_rewrite::rewrite_checked_int64_arith(logical_plan)?; // Sort-aware segment iteration. Mirror of `ContextIndexSearcher.shouldUseTimeSeriesDescSortOptimization` // for the indexed-parquet path. When the index has `index.sort.field` and the query's leading @@ -1408,6 +1415,8 @@ async unsafe fn execute_indexed_with_context_inner( ctx.register_table(®ister_name, provider)?; let logical_plan = from_substrait_plan(&ctx.state(), &plan).await?; + // Make BIGINT +,-,* overflow error instead of silently wrapping (see checked_arith_rewrite). + let logical_plan = crate::checked_arith_rewrite::rewrite_checked_int64_arith(logical_plan)?; log_debug!( "DataFusion logical plan:\n{}", logical_plan.display_indent() diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs index 8708627abe3a4..6558d676a3795 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs @@ -24,6 +24,7 @@ pub(crate) mod agg_mode; pub mod api; pub mod cache; pub mod cancellation; +pub mod checked_arith_rewrite; pub mod cross_rt_stream; pub mod datafusion_query_config; pub mod executor; diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/local_executor.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/local_executor.rs index 42bf70d811c4a..d07d8f6f97fbd 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/local_executor.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/local_executor.rs @@ -204,6 +204,8 @@ impl LocalSession { DataFusionError::Execution(format!("Failed to decode Substrait plan: {}", e)) })?; let logical_plan = from_substrait_plan(&self.ctx.state(), &plan).await?; + // Make BIGINT +,-,* overflow error instead of silently wrapping (see checked_arith_rewrite). + let logical_plan = crate::checked_arith_rewrite::rewrite_checked_int64_arith(logical_plan)?; log_debug!( "DataFusion logical plan:\n{}", logical_plan.display_indent() @@ -249,6 +251,8 @@ impl LocalSession { )) })?; let logical_plan = from_substrait_plan(&self.ctx.state(), &plan).await?; + // Make BIGINT +,-,* overflow error instead of silently wrapping (see checked_arith_rewrite). + let logical_plan = crate::checked_arith_rewrite::rewrite_checked_int64_arith(logical_plan)?; let dataframe = self.ctx.execute_logical_plan(logical_plan).await?; let physical_plan = dataframe.create_physical_plan().await?; // Strip first so `force_aggregate_mode(Final)` can find the Final/Partial pair diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs index ba2c9da1114f7..e7244cf2da04d 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs @@ -149,6 +149,8 @@ async fn build_dataframe( let substrait_plan = Plan::decode(plan_bytes) .map_err(|e| DataFusionError::Execution(format!("Failed to decode Substrait: {}", e)))?; let logical_plan = from_substrait_plan(&ctx.state(), &substrait_plan).await?; + // Make BIGINT +,-,* overflow error instead of silently wrapping (see checked_arith_rewrite). + let logical_plan = crate::checked_arith_rewrite::rewrite_checked_int64_arith(logical_plan)?; ctx.execute_logical_plan(logical_plan).await } @@ -251,6 +253,8 @@ pub async fn execute_with_context( // Union schema widening was applied at table registration (session_context::widen_to_union_schema). let logical_plan = from_substrait_plan(&handle.ctx.state(), &substrait_plan).await?; + // Make BIGINT +,-,* overflow error instead of silently wrapping (see checked_arith_rewrite). + let logical_plan = crate::checked_arith_rewrite::rewrite_checked_int64_arith(logical_plan)?; log_debug!( "DataFusion logical plan:\n{}", logical_plan.display_indent() diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/checked_arith.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/checked_arith.rs new file mode 100644 index 0000000000000..0c85ad0f12bd6 --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/checked_arith.rs @@ -0,0 +1,246 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +//! Overflow-checked BIGINT arithmetic UDFs: `checked_add_i64`, `checked_sub_i64`, +//! `checked_mul_i64`. +//! +//! DataFusion's default integer arithmetic (`Operator::Plus/Minus/Multiply`) uses arrow's +//! *wrapping* kernels, so `9223372036854775807 * 2` silently wraps into a negative i64 instead +//! of erroring. PPL callers expect an error, not a wrapped value. [`crate::checked_arith_rewrite`] +//! rewrites every `Int64 {+,-,*} Int64` logical `BinaryExpr` into a call to one of these UDFs, +//! which invoke arrow's *checked* numeric kernels (`add`/`sub`/`mul`) and surface a stable, +//! self-authored error on overflow. +//! +//! The overflow message carries [`OVERFLOW_KEYPHRASE`] verbatim; `NativeErrorConverter` (Java) +//! matches on that substring and maps it to an `IllegalArgumentException` (HTTP 400). + +use std::hash::{Hash, Hasher}; +use std::sync::Arc; + +use datafusion::arrow::array::ArrayRef; +use datafusion::arrow::compute::kernels::numeric::{add, mul, sub}; +use datafusion::arrow::datatypes::DataType; +use datafusion::common::{exec_datafusion_err, DataFusionError, Result}; +use datafusion::execution::context::SessionContext; +use datafusion::logical_expr::{ + ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility, +}; + +/// Stable Rust→Java contract phrase. `NativeErrorConverter.PATTERNS` matches on this exact +/// substring to classify the error as a client (400) error. Changing it requires a coordinated +/// edit in `NativeErrorConverter.java`. +pub const OVERFLOW_KEYPHRASE: &str = "BIGINT arithmetic overflow"; + +#[derive(Copy, Clone, Debug)] +enum Op { + Add, + Sub, + Mul, +} + +impl Op { + fn verb(self) -> &'static str { + match self { + Op::Add => "addition", + Op::Sub => "subtraction", + Op::Mul => "multiplication", + } + } +} + +/// `checked_{add,sub,mul}_i64(bigint, bigint) -> bigint`. Errors on 64-bit overflow instead of +/// wrapping. The return type is `Int64`, identical to the `BinaryExpr` it replaces, so the +/// rewrite leaves plan schemas byte-identical. +#[derive(Debug)] +pub struct CheckedArithUdf { + op: Op, + name: &'static str, + signature: Signature, +} + +impl CheckedArithUdf { + fn new(op: Op, name: &'static str) -> Self { + Self { + op, + name, + // Exact (Int64, Int64) — the rewrite only ever emits this shape. Immutable (not + // Volatile) so the UDF still participates in predicate pushdown and common-subexpression + // elimination. Const-folding an *overflowing* literal pair does not lose the error: + // DataFusion's ConstEvaluator preserves the original expr on a UDF runtime error + // (it only fails-fast for CAST), so overflow still surfaces at execution with our + // stable message; a non-overflowing constant simply folds to its exact value. + signature: Signature::exact( + vec![DataType::Int64, DataType::Int64], + Volatility::Immutable, + ), + } + } +} + +// `ScalarUDFImpl` requires DynEq + DynHash. Instances are distinguished only by name. +impl PartialEq for CheckedArithUdf { + fn eq(&self, other: &Self) -> bool { + self.name == other.name + } +} +impl Eq for CheckedArithUdf {} +impl Hash for CheckedArithUdf { + fn hash(&self, state: &mut H) { + self.name.hash(state); + } +} + +impl ScalarUDFImpl for CheckedArithUdf { + fn name(&self) -> &str { + self.name + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, _arg_types: &[DataType]) -> Result { + // Must equal the BinaryExpr output type (Int64) so plan schemas are unchanged by the + // rewrite (see checked_arith_rewrite: no recompute_schema). + Ok(DataType::Int64) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { + if args.args.len() != 2 { + return Err(exec_datafusion_err!( + "{} expects exactly 2 arguments, got {}", + self.name, + args.args.len() + )); + } + let n = args.number_rows; + // Normalize both operands to arrays; scalar-scalar calls arrive with number_rows == 1. + let lhs: ArrayRef = args.args[0].to_array(n)?; + let rhs: ArrayRef = args.args[1].to_array(n)?; + + // arrow's checked kernels null-propagate (null in either operand -> null out) and return + // Err(ArithmeticOverflow) only for non-null overflowing pairs. `&dyn Array` is a `Datum`. + let result = match self.op { + Op::Add => add(&lhs, &rhs), + Op::Sub => sub(&lhs, &rhs), + Op::Mul => mul(&lhs, &rhs), + }; + match result { + Ok(arr) => Ok(ColumnarValue::Array(arr)), + // Re-wrap arrow's ArithmeticOverflow into a stable, self-authored message so the + // Java-side keyphrase does not depend on arrow's wording across version bumps. + Err(e) => Err(DataFusionError::Execution(format!( + "{OVERFLOW_KEYPHRASE}: {} of two BIGINT values exceeded the 64-bit range \ + (underlying: {e})", + self.op.verb() + ))), + } + } +} + +fn make_udf(op: Op, name: &'static str) -> Arc { + Arc::new(ScalarUDF::from(CheckedArithUdf::new(op, name))) +} + +/// `checked_add_i64(bigint, bigint) -> bigint`. +pub fn checked_add_udf() -> Arc { + make_udf(Op::Add, "checked_add_i64") +} + +/// `checked_sub_i64(bigint, bigint) -> bigint`. +pub fn checked_sub_udf() -> Arc { + make_udf(Op::Sub, "checked_sub_i64") +} + +/// `checked_mul_i64(bigint, bigint) -> bigint`. +pub fn checked_mul_udf() -> Arc { + make_udf(Op::Mul, "checked_mul_i64") +} + +pub fn register_all(ctx: &SessionContext) { + ctx.register_udf(ScalarUDF::clone(&checked_add_udf())); + ctx.register_udf(ScalarUDF::clone(&checked_sub_udf())); + ctx.register_udf(ScalarUDF::clone(&checked_mul_udf())); +} + +#[cfg(test)] +mod tests { + use super::*; + use datafusion::arrow::array::{Array, AsArray, Int64Array}; + use datafusion::arrow::datatypes::{Field, Int64Type}; + use datafusion::logical_expr::ColumnarValue; + + fn call(udf: Arc, a: &[Option], b: &[Option]) -> Result { + let la: ArrayRef = Arc::new(Int64Array::from(a.to_vec())); + let lb: ArrayRef = Arc::new(Int64Array::from(b.to_vec())); + let return_field = Arc::new(Field::new("r", DataType::Int64, true)); + let args = ScalarFunctionArgs { + args: vec![ColumnarValue::Array(la), ColumnarValue::Array(lb)], + arg_fields: vec![ + Arc::new(Field::new("a", DataType::Int64, true)), + Arc::new(Field::new("b", DataType::Int64, true)), + ], + number_rows: a.len(), + return_field, + config_options: Arc::new(Default::default()), + }; + match udf.inner().invoke_with_args(args)? { + ColumnarValue::Array(arr) => Ok(arr), + other => panic!("expected array, got {other:?}"), + } + } + + #[test] + fn mul_overflow_errors_with_stable_phrase() { + let err = call(checked_mul_udf(), &[Some(i64::MAX)], &[Some(2)]).unwrap_err(); + assert!(err.to_string().contains(OVERFLOW_KEYPHRASE), "got: {err}"); + } + + #[test] + fn add_overflow_errors() { + let err = call(checked_add_udf(), &[Some(i64::MAX)], &[Some(1)]).unwrap_err(); + assert!(err.to_string().contains(OVERFLOW_KEYPHRASE), "got: {err}"); + } + + #[test] + fn sub_overflow_errors() { + let err = call(checked_sub_udf(), &[Some(i64::MIN)], &[Some(1)]).unwrap_err(); + assert!(err.to_string().contains(OVERFLOW_KEYPHRASE), "got: {err}"); + } + + #[test] + fn no_overflow_returns_exact_value() { + let out = call(checked_mul_udf(), &[Some(3)], &[Some(4)]).unwrap(); + assert_eq!(out.as_primitive::().value(0), 12); + } + + #[test] + fn negative_product_is_exact_not_overflow() { + // -3 * 20000 = -60000, well within i64 range; must not error. + let out = call(checked_mul_udf(), &[Some(-3)], &[Some(20000)]).unwrap(); + assert_eq!(out.as_primitive::().value(0), -60000); + } + + #[test] + fn null_operand_propagates_null_not_error() { + let out = call(checked_mul_udf(), &[None], &[Some(2)]).unwrap(); + assert!(out.as_primitive::().is_null(0)); + } + + #[test] + fn one_overflow_in_batch_fails_whole_call() { + // arrow's checked kernel errors on the first overflowing pair, failing the batch. + let err = call( + checked_mul_udf(), + &[Some(1), Some(i64::MAX)], + &[Some(1), Some(2)], + ) + .unwrap_err(); + assert!(err.to_string().contains(OVERFLOW_KEYPHRASE), "got: {err}"); + } +} diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs index 853b85fe13cd5..c68fb4eadfce1 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs @@ -121,6 +121,7 @@ pub(crate) fn coerce_args( } pub mod binary_to_base64; +pub mod checked_arith; pub mod conv; pub mod conversion; pub mod convert_tz; @@ -175,6 +176,7 @@ pub mod width_bucket; // and restart the OpenSearch JVM (the loaded dylib is JVM-cached). pub fn register_all(ctx: &SessionContext) { binary_to_base64::register_all(ctx); + checked_arith::register_all(ctx); conv::register_all(ctx); convert_tz::register_all(ctx); conversion::register_all(ctx); diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/NativeErrorConverter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/NativeErrorConverter.java index ccabcf0e86fb5..68485ea82240c 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/NativeErrorConverter.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/NativeErrorConverter.java @@ -66,6 +66,9 @@ private record MatchedError(String message, Exception original) { private static final String ADMISSION_REJECTED_MSG = "Native query admission rejected: insufficient memory budget available"; + /** Stable Rust→Java contract phrase; must match {@code OVERFLOW_KEYPHRASE} in udf/checked_arith.rs. */ + static final String BIGINT_OVERFLOW_MSG = "BIGINT arithmetic overflow"; + /** * Registered error patterns, checked in order. First match wins. * Key phrases include both the Rust-originated messages (for data-node local conversion) @@ -92,7 +95,11 @@ private record MatchedError(String message, Exception original) { // (deeply nested function calls). Converted at the FFM boundary into a clean 400. new ErrorPattern("recursion limit reached", NativeErrorConverter::convertRecursionLimit), // Controlled message, as a coordinator-side safety net if it arrives via StreamException. - new ErrorPattern("Query too deeply nested", NativeErrorConverter::convertRecursionLimit) + new ErrorPattern("Query too deeply nested", NativeErrorConverter::convertRecursionLimit), + // BIGINT (Int64) +, -, * overflow produced by the checked_arith UDF (see the Rust + // udf/checked_arith.rs). Client error → HTTP 400. The message is self-authored (carries no + // native internals), so it is surfaced verbatim from the keyphrase onward. + new ErrorPattern(BIGINT_OVERFLOW_MSG, NativeErrorConverter::convertBigintOverflow) ); /** @@ -172,6 +179,23 @@ private static Exception convertRecursionLimit(MatchedError match) { ); } + private static Exception convertBigintOverflow(MatchedError match) { + // The message is self-authored by the Rust UDF and carries no native internals, so surface it + // from the keyphrase onward — stripping any DataFusion "Execution error:" wrapper prefix and + // trailing cause detail. No cause is attached (parity with the other converters). + // + // Return a status-bearing OpenSearchStatusException (400) rather than a bare + // IllegalArgumentException: on distributed-aggregate paths the failure is re-wrapped several + // times across the Flight boundary (e.g. RuntimeException("Failed to start streaming + // fragment", ...) -> StreamException), and only a typed OpenSearchException carrying a + // non-500 status is recovered by the coordinator's statusBearingCause walk. A plain + // IllegalArgumentException would be redacted to a generic 500 there. + String msg = match.message(); + int start = msg.indexOf(BIGINT_OVERFLOW_MSG); + String clean = start >= 0 ? msg.substring(start) : BIGINT_OVERFLOW_MSG; + return new OpenSearchStatusException(clean, RestStatus.BAD_REQUEST); + } + // ─── Message parsing ──────────────────────────────────────────────────────── /** diff --git a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/NativeErrorConverterTests.java b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/NativeErrorConverterTests.java index bd985222bd28b..ffd11bad003f5 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/NativeErrorConverterTests.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/NativeErrorConverterTests.java @@ -159,6 +159,36 @@ public void testControlledRecursionMessageConvertsOnCoordinator() { assertTrue(result.getMessage().contains("too deeply nested")); } + public void testBigintOverflowConvertsToBadRequestStatusException() { + // Raw message as authored by the Rust checked_arith UDF, wrapped by DataFusion's Display. + String message = "Execution error: BIGINT arithmetic overflow: multiplication of two BIGINT values " + + "exceeded the 64-bit range (underlying: Arithmetic overflow: Overflow happened on: 9223372036854775807 * 2)"; + RuntimeException original = new RuntimeException(message); + + Exception result = NativeErrorConverter.convert(original); + + // Status-bearing 400 so distributed-aggregate re-wrapping still surfaces a client error. + assertTrue(result instanceof OpenSearchStatusException); + assertEquals(RestStatus.BAD_REQUEST, ((OpenSearchStatusException) result).status()); + // Message is surfaced from the keyphrase onward; the "Execution error:" wrapper is stripped. + assertTrue(result.getMessage().startsWith("BIGINT arithmetic overflow")); + assertTrue(result.getMessage().contains("exceeded the 64-bit range")); + // No cause attached (parity with the other converters). + assertNull(result.getCause()); + } + + public void testBigintOverflowControlledMessageConvertsOnCoordinator() { + // Coordinator-side safety net: the message arriving via StreamException after transport. + String controlledMessage = "BIGINT arithmetic overflow: addition of two BIGINT values exceeded the 64-bit range"; + RuntimeException transportException = new RuntimeException(controlledMessage); + + Exception result = NativeErrorConverter.convert(transportException); + + assertTrue(result instanceof OpenSearchStatusException); + assertEquals(RestStatus.BAD_REQUEST, ((OpenSearchStatusException) result).status()); + assertEquals(controlledMessage, result.getMessage()); + } + public void testUnrecognizedErrorPassedThrough() { RuntimeException original = new RuntimeException("Some unknown error from native code"); diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/AnalyticsTransportErrors.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/AnalyticsTransportErrors.java index 38ada4e8f848b..1b9819c1c2056 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/AnalyticsTransportErrors.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/AnalyticsTransportErrors.java @@ -66,9 +66,29 @@ static Exception toWireError(Exception e) { if (cancelled != null) { return new StreamException(StreamErrorCode.CANCELLED, cancelled.getMessage(), e); } + // A BIGINT-arithmetic overflow is a client error (HTTP 400): the data node's + // NativeErrorConverter has already turned it into an IllegalArgumentException carrying the + // stable phrase. Without an explicit code it would cross Flight as INTERNAL and surface as a + // generic 500 on distributed-aggregate paths. Tag it INVALID_ARGUMENT so fromWireError can + // rebuild a 400 on the coordinator. + Throwable badRequest = ExceptionsHelper.unwrapCausesAndSuppressed( + e, + t -> t.getMessage() != null && t.getMessage().contains(BIGINT_OVERFLOW_PHRASE) + ).orElse(null); + if (badRequest != null) { + return new StreamException(StreamErrorCode.INVALID_ARGUMENT, badRequest.getMessage(), e); + } return e; } + /** + * Stable phrase identifying a BIGINT-arithmetic overflow client error, produced by the + * DataFusion backend's {@code checked_arith} UDF and surfaced by {@code NativeErrorConverter}. + * Kept in sync with {@code NativeErrorConverter.BIGINT_OVERFLOW_MSG} and the Rust + * {@code OVERFLOW_KEYPHRASE}. + */ + static final String BIGINT_OVERFLOW_PHRASE = "BIGINT arithmetic overflow"; + /** * Coordinator receive side. Inverse of {@link #toWireError}: maps a {@link StreamException} that * crossed transport back to a typed exception, so a shard-side failure doesn't surface to the user @@ -80,6 +100,8 @@ static Exception toWireError(Exception e) { * drop (node gone, connection reset) is "service unavailable", not an internal error. *
  • {@link StreamErrorCode#CANCELLED} → {@link TaskCancelledException} — a shard cancel (e.g. SBP) * stays a recognizable cancellation instead of degrading to a generic ISE that clients would retry. + *
  • {@link StreamErrorCode#INVALID_ARGUMENT} → 400 {@link OpenSearchStatusException} — a client + * error (e.g. BIGINT arithmetic overflow) stays a 400 instead of being redacted to a generic 500. * * Anything else passes through unchanged. */ @@ -89,12 +111,17 @@ static Exception fromWireError(Exception e) { t -> t instanceof StreamException s && (s.getErrorCode() == StreamErrorCode.RESOURCE_EXHAUSTED || s.getErrorCode() == StreamErrorCode.UNAVAILABLE - || s.getErrorCode() == StreamErrorCode.CANCELLED) + || s.getErrorCode() == StreamErrorCode.CANCELLED + || s.getErrorCode() == StreamErrorCode.INVALID_ARGUMENT) ).orElse(null); if (se == null) { return e; } String message = se.getMessage(); + if (se.getErrorCode() == StreamErrorCode.INVALID_ARGUMENT) { + // Client error (e.g. BIGINT arithmetic overflow) → 400, so it isn't redacted to a 500. + return new OpenSearchStatusException(message != null ? message : "invalid argument", RestStatus.BAD_REQUEST, e); + } if (se.getErrorCode() == StreamErrorCode.RESOURCE_EXHAUSTED) { // CircuitBreakingException has no cause-accepting ctor; attach the wire exception via initCause // so the original StreamException/stack is kept for server-side troubleshooting. diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/AnalyticsTransportErrorsTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/AnalyticsTransportErrorsTests.java index 13e4f0153bb99..46465515cefc7 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/AnalyticsTransportErrorsTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/AnalyticsTransportErrorsTests.java @@ -165,4 +165,66 @@ public void testTaskCancelledSurvivesRoundTripAsTaskCancelled() { ); assertEquals("sbp cancel", recovered.getMessage()); } + + // ── BIGINT overflow (client 400) ────────────────────────────────────── + + public void testToWireErrorTagsBigintOverflowAsInvalidArgument() { + // NativeErrorConverter turns the DataFusion overflow into this IllegalArgumentException. + IllegalArgumentException overflow = new IllegalArgumentException( + "BIGINT arithmetic overflow: multiplication of two BIGINT values exceeded the 64-bit range" + ); + + Exception wire = AnalyticsTransportErrors.toWireError(overflow); + + assertTrue(wire instanceof StreamException); + assertEquals(StreamErrorCode.INVALID_ARGUMENT, ((StreamException) wire).getErrorCode()); + assertTrue(wire.getMessage().contains("BIGINT arithmetic overflow")); + } + + public void testToWireErrorFindsBigintOverflowInCauseChain() { + // The distributed-aggregate path wraps the failure: RuntimeException("Stage N failed", cause). + Exception wrapped = new RuntimeException( + "Stage 0 failed", + new IllegalArgumentException("BIGINT arithmetic overflow: addition of two BIGINT values exceeded the 64-bit range") + ); + + Exception wire = AnalyticsTransportErrors.toWireError(wrapped); + + assertTrue(wire instanceof StreamException); + assertEquals(StreamErrorCode.INVALID_ARGUMENT, ((StreamException) wire).getErrorCode()); + } + + public void testFromWireErrorRebuildsBigintOverflowAs400() { + StreamException wire = new StreamException(StreamErrorCode.INVALID_ARGUMENT, "BIGINT arithmetic overflow: ..."); + + Exception recovered = AnalyticsTransportErrors.fromWireError(wire); + + assertTrue(recovered instanceof OpenSearchStatusException); + assertEquals(RestStatus.BAD_REQUEST, ((OpenSearchStatusException) recovered).status()); + assertTrue(recovered.getMessage().contains("BIGINT arithmetic overflow")); + } + + public void testFromWireErrorFindsInvalidArgumentInCauseChain() { + Exception wrapped = new RuntimeException( + "Stage 0 failed", + new StreamException(StreamErrorCode.INVALID_ARGUMENT, "BIGINT arithmetic overflow: ...") + ); + + Exception recovered = AnalyticsTransportErrors.fromWireError(wrapped); + + assertTrue(recovered instanceof OpenSearchStatusException); + assertEquals(RestStatus.BAD_REQUEST, ((OpenSearchStatusException) recovered).status()); + } + + public void testBigintOverflowSurvivesRoundTripAs400() { + IllegalArgumentException overflow = new IllegalArgumentException( + "BIGINT arithmetic overflow: multiplication of two BIGINT values exceeded the 64-bit range" + ); + + Exception onWire = AnalyticsTransportErrors.toWireError(overflow); + Exception recovered = AnalyticsTransportErrors.fromWireError(onWire); + + assertTrue(recovered instanceof OpenSearchStatusException); + assertEquals(RestStatus.BAD_REQUEST, ((OpenSearchStatusException) recovered).status()); + } } diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/BigintOverflowIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/BigintOverflowIT.java new file mode 100644 index 0000000000000..8f419549ad4de --- /dev/null +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/BigintOverflowIT.java @@ -0,0 +1,157 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.analytics.qa; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; + +import org.opensearch.client.Request; +import org.opensearch.client.ResponseException; + +/** + * Verifies that BIGINT (Int64) {@code +}, {@code -}, {@code *} overflow on the analytics-engine + * DataFusion backend returns an HTTP 400 error instead of silently wrapping to a negative value. + * + *

    DataFusion's default integer arithmetic uses arrow's wrapping kernels; the + * {@code checked_arith_rewrite} logical pass rewrites {@code Int64 op Int64} arithmetic to the + * overflow-checked {@code checked_*_i64} UDFs, whose error is mapped to 400 by + * {@code NativeErrorConverter}. Non-overflowing arithmetic and floating-point arithmetic are + * unaffected. + * + *

    The {@code long_val * long_val} overflow mirrors the reported large-multiplier case where a + * product exceeds the signed 64-bit range. + */ +public class BigintOverflowIT extends AnalyticsRestTestCase { + + private static final String INDEX = "bigint_overflow_test"; + private boolean provisioned = false; + + private void ensureProvisioned() throws IOException { + if (provisioned) { + return; + } + try { + client().performRequest(new Request("DELETE", "/" + INDEX)); + } catch (ResponseException e) { + if (e.getResponse().getStatusLine().getStatusCode() != 404) { + throw e; + } + } + + Request create = new Request("PUT", "/" + INDEX); + create.setJsonEntity( + "{" + + "\"settings\": {" + + " \"index.number_of_shards\": 1," + + " \"index.number_of_replicas\": 0," + + " \"index.pluggable.dataformat.enabled\": true," + + " \"index.pluggable.dataformat\": \"composite\"," + + " \"index.composite.primary_data_format\": \"parquet\"," + + " \"index.composite.secondary_data_formats\": [\"lucene\"]" + + "}," + + "\"mappings\": {" + + " \"properties\": {" + + " \"long_val\": {\"type\": \"long\"}," + + " \"double_val\": {\"type\": \"double\"}" + + " }" + + "}" + + "}" + ); + client().performRequest(create); + + Request bulk = new Request("POST", "/" + INDEX + "/_bulk"); + bulk.addParameter("refresh", "true"); + // 5000000000000000000 ≈ 5e18; times 2 overflows i64 max (~9.22e18). times 1 does not. + bulk.setJsonEntity( + "{\"index\":{}}\n{\"long_val\":5000000000000000000,\"double_val\":3.5}\n" + + "{\"index\":{}}\n{\"long_val\":3,\"double_val\":8.2}\n" + ); + client().performRequest(bulk); + + Request flush = new Request("POST", "/" + INDEX + "/_flush"); + flush.addParameter("force", "true"); + client().performRequest(flush); + + Request health = new Request("GET", "/_cluster/health/" + INDEX); + health.addParameter("wait_for_status", "yellow"); + health.addParameter("timeout", "30s"); + client().performRequest(health); + + provisioned = true; + } + + public void testLongColumnMultiplyOverflowReturns400() throws IOException { + ensureProvisioned(); + // long_val (5e18) * 2 overflows i64 for the first row. Column arithmetic reaches the + // DataFusion backend (a pure literal * literal would be constant-folded by the PPL/Calcite + // front end before ever reaching the backend, so we always overflow via a column operand). + assertOverflow400("source=" + INDEX + " | eval x = long_val * 2 | fields x"); + } + + public void testLongLiteralPlusColumnOverflowReturns400() throws IOException { + ensureProvisioned(); + // i64::MAX + long_val overflows for the first (non-zero) row. + assertOverflow400("source=" + INDEX + " | eval x = 9223372036854775807 + long_val | fields x"); + } + + public void testArithmeticInWherePredicateOverflowReturns400() throws IOException { + ensureProvisioned(); + // Arithmetic inside a WHERE predicate must also be checked. On the indexed path this + // predicate seeds the parquet pushdown; without rewriting the extraction plan it would be + // scan-filtered with wrapping arithmetic and drop the overflowing row silently. + assertOverflow400("source=" + INDEX + " | where long_val * 2 > 0 | fields long_val"); + } + + public void testArithmeticInStatsAggregateOverflowReturns400() throws IOException { + ensureProvisioned(); + // Arithmetic inside an aggregate argument must be reached (sum over the per-row product, + // where long_val * long_val overflows for the 5e18 row). + assertOverflow400("source=" + INDEX + " | stats sum(long_val * long_val) as s | fields s"); + } + + public void testNonOverflowingLongArithmeticSucceeds() throws IOException { + ensureProvisioned(); + // 3 * 4 = 12, well within range: must return 200 with the correct value. + Map response = executePpl("source=" + INDEX + " | where long_val = 3 | eval x = long_val * 4 | fields x"); + @SuppressWarnings("unchecked") + List> rows = (List>) response.get("datarows"); + assertNotNull(rows); + assertEquals(1, rows.size()); + assertEquals(12L, ((Number) rows.get(0).get(0)).longValue()); + } + + public void testDoubleArithmeticDoesNotError() throws IOException { + ensureProvisioned(); + // Floating-point arithmetic never routes through the checked rewrite (gate is Int64-only); + // it must return 200 — double overflow wraps to Infinity by PPL/Calcite parity, never 400. + Map response = executePpl( + "source=" + INDEX + " | where long_val = 3 | eval x = double_val * double_val | fields x" + ); + @SuppressWarnings("unchecked") + List> rows = (List>) response.get("datarows"); + assertNotNull(rows); + assertEquals(1, rows.size()); + } + + private void assertOverflow400(String ppl) throws IOException { + Request request = new Request("POST", "/_analytics/ppl"); + request.setJsonEntity("{\"query\": \"" + escapeJson(ppl) + "\"}"); + try { + client().performRequest(request); + fail("Expected 400 BIGINT overflow error for query: " + ppl); + } catch (ResponseException e) { + int status = e.getResponse().getStatusLine().getStatusCode(); + String body = new String(e.getResponse().getEntity().getContent().readAllBytes(), StandardCharsets.UTF_8); + assertEquals("Expected 400 but got " + status + ": " + body, 400, status); + assertTrue("Error should mention BIGINT overflow, got: " + body, body.contains("BIGINT arithmetic overflow")); + } + } +} From ea95017c925ded00a5df658a959e1bfbcf7aaa36 Mon Sep 17 00:00:00 2001 From: Kai Huang Date: Wed, 8 Jul 2026 10:26:45 -0700 Subject: [PATCH 2/2] Fix flaky NativeArenaPurgerTests.testPurgeOnlyFiresAboveThreshold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The purge thread runs on a fixed interval (200ms, set in setUp via init(0, 200) with threshold=0 = always purge). testPurgeOnlyFiresAboveThreshold raised the threshold to Long.MAX_VALUE and immediately sampled the purge count, but a purge cycle already in flight — started while the threshold was still 0, before the Rust thread observed the new threshold over FFI — could still fire once, yielding "expected:<0> but was:<1>". Wait past one full check interval after raising the threshold, before sampling the baseline count, so the thread finishes any in-flight cycle and observes the new threshold first. This mirrors the existing guard in testPurgeDisabledWhenIntervalIsZero. Test-only change; no production behavior change. Signed-off-by: Kai Huang --- .../opensearch/nativebridge/spi/NativeArenaPurgerTests.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/sandbox/libs/dataformat-native/src/test/java/org/opensearch/nativebridge/spi/NativeArenaPurgerTests.java b/sandbox/libs/dataformat-native/src/test/java/org/opensearch/nativebridge/spi/NativeArenaPurgerTests.java index c174e9c7c8ed4..ccee9f0ff11bc 100644 --- a/sandbox/libs/dataformat-native/src/test/java/org/opensearch/nativebridge/spi/NativeArenaPurgerTests.java +++ b/sandbox/libs/dataformat-native/src/test/java/org/opensearch/nativebridge/spi/NativeArenaPurgerTests.java @@ -53,6 +53,11 @@ public void testPurgeFiresPeriodically() throws Exception { public void testPurgeOnlyFiresAboveThreshold() throws Exception { // Set threshold very high — purge should never fire NativeArenaPurger.setThresholdBytes(Long.MAX_VALUE); + // Wait past one full check interval so the purge thread finishes any in-flight cycle + // (started while the threshold was still 0) and observes the new threshold before we + // sample the baseline count. Mirrors the guard in testPurgeDisabledWhenIntervalIsZero. + Thread.sleep(500); + long before = NativeArenaPurger.getPurgeCount(); Thread.sleep(500); assertEquals("Purge should not fire when below threshold", before, NativeArenaPurger.getPurgeCount());