Feature/datanode profiling v2#8
Draft
finnegancarroll wants to merge 96 commits into
Draft
Conversation
…s in queue that are cancelled (opensearch-project#21862) * Adding grpc interceptor to help with clearing up arrow buffer messages in queue that are cancelled Signed-off-by: Harsha Vamsi Kalluri <harshavamsi096@gmail.com> * Use explicit charset in test marshaller to satisfy forbiddenApis Signed-off-by: Harsha Vamsi Kalluri <harshavamsi096@gmail.com> --------- Signed-off-by: Harsha Vamsi Kalluri <harshavamsi096@gmail.com>
…20) (opensearch-project#21748) On the analytics-engine route, the SQL plugin wraps every datetime root column in `CAST(<DATE/TIME/TIMESTAMP> AS VARCHAR)`, and this rewriter translates those casts into DataFusion's `to_char` extension. Whenever the rewriter's format string and the PPL formatter disagree (e.g. trailing `Z`, `T` separator), users see wire-format divergence — opensearch-project/sql#5420. Let the analytics engine return real datetime cells. The companion PR in `opensearch-project/sql` removes the cast rule. The PPL response pipeline already handles datetime → string conversion natively at the formatter layer (`ExprTimestampValue.value()` etc.), so no engine-side formatting is needed. - Delete `DatetimeOutputCastRewriter` and its tests. - Remove the two `convertFragment` / `convertStandalone` callsites in `DataFusionFragmentConvertor`. - Drop the test that asserted `to_char` extension was emitted from `CAST(... VARCHAR)`. - Strip stale doc comments referencing the rewriter. - Keep the `TO_CHAR -> to_char` function mapping in `opensearch_scalar_functions.yaml` for any unrelated paths that may still emit `TO_CHAR` directly. Signed-off-by: Eric Wei <mengwei.eric@gmail.com>
…e analytics engine planner see the same ClusterState (opensearch-project#21873) * move EngineContext to analytics-api Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * [analytics-engine] EngineContext returns QueryEngineContext directly - Drop getSchema/operatorTable; expose getContext() and getContext(state) - Type QueryPlanExecutor's context param as QueryEngineContext - Update internal callers to queryCtx.schema() Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * test updates Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * pr feedback Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * spotless Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * fix merge conflict Signed-off-by: Marc Handalian <marc.handalian@gmail.com> --------- Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…rch-project#21866) * analytics-engine: replica failover on shard dispatch failure Wire ShardFragmentStageExecution.retargetForRetry to advance the failed task's ShardExecutionTarget through the underlying ShardIterator, mirroring AbstractSearchAsyncAction.onShardFailure → FailAwareWeightedRouting.findNext. ShardTargetResolver now hands the post-initial-pop iterator + ClusterState into the target so failover state travels with the task; subsequent nextCopy() invocations resolve the next replica's node from cluster state and produce a fresh target for the retry. Any dispatch failure advances the iterator; eligibility is enforced by exhaustion rather than exception-type filtering, matching the search API's no-filter semantics. Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * spotless Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * fix merge conflict Signed-off-by: Marc Handalian <marc.handalian@gmail.com> --------- Signed-off-by: Marc Handalian <marc.handalian@gmail.com> Co-authored-by: Sandesh Kumar <sandeshkr419@gmail.com>
* Refactor distributed-aggregate rewriter - Decompose DistributedAggregateRewriter into a transformer pipeline (FinalInputTransformer + ExchangeTypeOverride / LiteralProject transformers + thin orchestrator). - Own construction-time aggCall rebase + COUNT→SUM swap + type pinning in FinalAggCallBuilder (planner.rel); split rule no longer reaches across packages. - Classify each call's IntermediateField once on the ORIGINAL aggCalls and stash via CallDecomposition record — post-Volcano consumers never re-classify post-swap calls. - OpenSearchAggregateSplitRule now builds FINAL via FinalAggCallBuilder with the stashed classification, fixing aggCall slot adjustment when the input schema changes during the aggregate split. - Covered unit tests for DistributedAggregateRewriterTests and AggregatePlanShapeTests. Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> * Updated comments Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> --------- Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
…21872) Signed-off-by: rayshrey <rayshrey@amazon.com>
… Foyer key_index persistence, and fix retention lease renewal (opensearch-project#21849) * Add foyer file based registry Signed-off-by: Vishwas Garg <vishwasgarg14@gmail.com> * handle renew retention lease, handle cancel hot to warm write blocks, update block cache settings and misc changes Signed-off-by: Vishwas Garg <vishwasgarg14@gmail.com> * Skip update the write block removal check if cluster metadata is updated. write block applied in prepare action triggers remove write Signed-off-by: Vishwas Garg <vishwasgarg14@gmail.com> --------- Signed-off-by: Vishwas Garg <vishwasgarg14@gmail.com>
…earch-project#21870) FlightTransportTestBase was assigning a single deterministic port per test via getBasePort(9500) + portCounter. When that port was held in TIME_WAIT or otherwise contended on a CI runner, the FlightTransport bind failed with BindTransportException, causing flakes in FlightOutboundHandlerContextPropagationTests, FlightClientChannelTests, and FlightMetricsTests. Pass a 100-port range to the aux.transport.transport-flight.port setting so PortsRange.iterate() finds the first free port, and read the actual bound port back from FlightTransport.boundAddress() to construct the test's stream and transport addresses. This matches the pattern used by AbstractSimpleTransportTestCase, MockTransportService, and OpenSearchSingleNodeTestCase. Signed-off-by: Andrew Ross <andrross@amazon.com>
- Bump jetty 9.4.57 -> 9.4.58 in hdfs-fixture (CVE-2026-2332, latest public 9.4.x, no full fix available for EOL line) - Bump kafka-clients 3.9.1 -> 3.9.2 (CVE-2026-35554) - Upgrade maven-model 3.9.12 -> 3.9.16, force plexus-utils 4.0.3 (CVE-2025-67030) - Force log4j-core to 2.25.4 in buildSrc (CVE-2026-34480, CVE-2026-34478, CVE-2026-34477) Signed-off-by: Craig Perkins <cwperx@amazon.com>
… queries (opensearch-project#21775) Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>
…#21901) These four tests were muted by opensearch-project#21774 right after opensearch-project#21731 introduced them. The underlying aggCall slot adjustment / IntermediateField re-classification issue was fixed by opensearch-project#21875 (Refactor distributed-aggregate rewriter). Verified locally: 12/12 runs pass with -Dtests.iters=3. Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>
… shim (opensearch-project#21894) * update sandbox tests to conditionally use ppl endoint. There is too much variance between the real sql plugin planner and test-ppl-frontend plugin. This wires up the real endpoint for our qa package tests, and mutes tests awaiting fixes. Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * mute failing tests Signed-off-by: Marc Handalian <marc.handalian@gmail.com> --------- Signed-off-by: Marc Handalian <marc.handalian@gmail.com> Co-authored-by: Sandesh Kumar <sandeshkr419@gmail.com>
…pensearch-project#21874) Populates the Log4j2 ThreadContext (MDC) with the X-Opaque-Id header value on both the coordinator and data node execution threads. This allows operators to trace analytics engine queries through planning, scheduling, and execution logs by adding %X{opaque_id} to their log4j2.properties pattern. Coordinator: MDC is set on the SEARCH thread in DefaultPlanExecutor before planning/execution begins, and cleared in finally. Data node: MDC is set on the SEARCH thread inside the wrapped executor in AnalyticsSearchTransportService's fragment execution handler. Signed-off-by: carrofin <carrofin@amazon.com> Signed-off-by: Finn Carroll <carrofin@amazon.com>
…ons (opensearch-project#21902) The delegation_possible / index_filter marker UDFs were registered only on shard scans. A coordinator-reduce session has a step to parse a producer (shard) plan to derive its output schema and failed with "Unsupported function name: delegation_possible". Register both markers in udf::register_all so every session can parse them; only the indexed data-node path ever unwraps/executes them. Un-mutes JoinCommandIT left/right-outer + left-anti join tests. Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…apes (opensearch-project#21804) * Add subquery ITs and relax OpenSearchJoinRule for mixed conditions Builds on PR opensearch-project#21795 (Decorrelate RexSubQuery before marking). ITs ported from opensearch-project/sql: * InSubqueryCommandIT (13 tests) * ScalarSubqueryCommandIT (13 tests) * ExistsSubqueryCommandIT (11 tests) * worker / work_information / occupation datasets OpenSearchJoinRule.matches() relaxed to accept mixed equi+non-equi conditions, IS_NOT_DISTINCT_FROM (null-safe equi), and equi conjuncts where one operand is a RexCall expression (e.g. uid = salary + 1000) - DataFusion hash-join evaluates each operand to derive the key. RelDecorrelator's output for streamstats reset's directly-built LogicalCorrelate carries an AND(<=, =, =) condition; without this relaxation the rule rejected it. StreamstatsCommandIT.testStreamstatsReset flipped to positive (the other two reset_* variants are flipped in the follow-up commit that adds LITERAL_AGG lowering). Signed-off-by: Songkan Tang <songkant@amazon.com> * Lower LITERAL_AGG and accept all join condition shapes in marking Two complementary marking-phase fixes layered on top of PR opensearch-project#21795, both keeping Calcite as the source of truth: 1. PlannerImpl.removeSubQueries — after RelDecorrelator, run a Hep pass with ExtractLiteralAggRule (ported from Apache Impala) to lower LITERAL_AGG(literal) aggregate calls into Aggregate + Project. SubQueryRemoveRule.rewriteIn emits LITERAL_AGG for NOT IN's null-aware semantics; Calcite has no rule to remove it; each downstream engine handles it differently (Druid registers a native impl, Ignite wires an accumulator, Impala lowers to Aggregate + Project, Flink forks SubQueryRemoveRule). The Impala approach is a clean Calcite-rule-style rewrite — port it with attribution. 2. OpenSearchJoinRule.matches() — accept all join condition shapes, not just equi or mixed equi+non-equi. RelDecorrelator's output for correlated subqueries with non-equi correlation predicates (e.g. WHERE id > uid) is a pure non-equi LogicalJoin. DataFusion handles non-equi via NestedLoopJoin downstream, so there is no capability-level reason to reject these at the marking phase. End-to-end IT results on this branch (PR opensearch-project#21795 + these two fixes): ExistsSubqueryCommandIT : 11/11 InSubqueryCommandIT : 11/13 (1 row-order, 1 unrelated AggregateSplitRule bug) ScalarSubqueryCommandIT : 13/13 StreamstatsCommandIT : testStreamstatsReset passes Total : 36/38 = 95% The two remaining failures are unrelated to subquery decorrelation: * testTwoExpressionsInSubquery — result row ordering between hash join's bucketing and the IT's exact-row assertion. Test issue. * testTwoExpressionsNotInSubquery — OpenSearchAggregateSplitRule rejects nullable filter expressions ("filter must be BOOLEAN NOT NULL") on a multi-column NOT IN's COUNT FILTER. Independent AggregateSplitRule bug. Signed-off-by: Songkan Tang <songkant@amazon.com> * Reconcile streamstats + InSubquery ITs with newly-supported plans CI surfaced four StreamstatsCommandIT failures and two InSubqueryCommandIT failures. All six are reconciled here. Streamstats — testStreamstatsResetWithNull / GlobalWithNull / ResetWithNullBucket / GlobalWithNullBucket previously asserted the query *must error* via assertErrorAny. After PR opensearch-project#21795's decorrelation plus this branch's marking-phase fixes (LITERAL_AGG lowering and OpenSearchJoinRule relaxation), these PPL forms now plan and execute end-to-end. Flip the four assertions to executePpl + assertNotNull, matching how testStreamstatsReset is already structured. InSubquery — testTwoExpressionsInSubquery's exact-row assertion was fragile across hash-join output ordering for two rows sharing salary=120000. Add 'id' as a secondary sort key so the order is fully determined. InSubquery — testTwoExpressionsNotInSubquery hits an independent OpenSearchAggregateSplitRule bug ("filter must be BOOLEAN NOT NULL") unrelated to subquery decorrelation. Mark @AwaitsFix with a note pointing at the split-rule limitation. Local verification: * StreamstatsCommandIT — all tests pass. * InSubqueryCommandIT — 12/13 pass, the @AwaitsFix-skipped one is the split-rule case. Signed-off-by: Songkan Tang <songkant@amazon.com> * Trim subquery ITs and simplify OpenSearchJoinRule.matches Two scope reductions following review feedback: * Remove ExistsSubqueryCommandIT, InSubqueryCommandIT, ScalarSubqueryCommandIT and their worker / work_information / occupation datasets. These are subquery-decorrelation coverage ports from the upstream sql repo and don't directly verify the marking-phase fixes layered here. They can be re-added in a dedicated subquery-IT PR if/when the analytics-engine route wants to track that suite. * Simplify OpenSearchJoinRule.matches() to a plain JoinRelType whitelist (INNER / LEFT / RIGHT / FULL / SEMI / ANTI). The condition-shape acceptance is implicit — DataFusion's join planner picks the physical strategy (HashJoin / NestedLoopJoin / CrossJoin) downstream, so there is no need for the rule to inspect the condition. JoinCommandIT.testAppendcol is no longer @AwaitsFix — appendcol's ROW_NUMBER + FULL OUTER LogicalJoin lowering now planes and executes end-to-end on calcs (12/12 JoinCommandIT tests green locally). Signed-off-by: Songkan Tang <songkant@amazon.com> * Address review comments * PlannerImpl.runAllOptimizations: drop the post-removeSubQueries LOGGER.info dump — the per-phase RuleProfilingListener output already covers what the log line gave. * PlannerImpl.extractLiteralAgg: import ExtractLiteralAggRule and use the simple class name at the call site (and in the @link reference in javadoc). * ExtractLiteralAggRule: import java.util.Objects and use Objects::nonNull instead of the fully-qualified java.util.Objects::nonNull. Signed-off-by: Songkan Tang <songkant@amazon.com> * HepPhase: distinguish RuleInstance from RuleCollection The previous HepPhase API exposed only addRules(...) and chose addRuleInstance vs addRuleCollection at run() time based on whether the accumulated list had one element or many. That collapsed two Hep instructions with different fixpoint semantics into one branch: * addRuleInstance(R) appends a HepInstruction.RuleInstance — R runs to fixpoint, then the next instruction starts. * addRuleCollection([R1, R2, ...]) appends a RuleCollection — every rule in the group is tried at each vertex within the same fixpoint loop, so a rewrite by one rule can cascade into a match by another in the same pass. The two are not interchangeable. PlannerImpl on main already relied on the distinction in pushdownRules: three transposes were grouped in a RuleCollection so they cascade-converge, then FILTER_MERGE was a separate RuleInstance so it only fired after the transposes settled (otherwise auto-injected NOT NULL would merge with a half- pushed intermediate filter rather than the user's WHERE on the post-pushdown shape). This commit: * Renames the HepPhase API to addRuleInstance / addRuleCollection, mirroring HepProgramBuilder one-to-one. addRuleCollection takes Collection<? extends RelOptRule>. * Stores instructions as List<Consumer<HepProgramBuilder>> and applies them in declared order at run() — no more size-based branching. * Restores PlannerImpl call sites to the original instruction structure: pushdownRules splits back into a transpose collection + a FILTER_MERGE instance, and the single-rule phases use addRuleInstance. Local verification: analytics-engine unit tests, JoinCommandIT, StreamstatsCommandIT all green. Signed-off-by: Songkan Tang <songkant@amazon.com> --------- Signed-off-by: Songkan Tang <songkant@amazon.com>
* fix(datafusion): complete coordinator reduce cancellation 1. AbortHandle not registered for reduce paths — execute_local_plan and execute_local_prepared_plan used CrossRtStream::new_with_df_error_stream which discards the AbortHandle. Switch both to the cancellable variant and call set_abort_handle so the CPU executor task is aborted when cancel_query fires, not just at the next stream_next boundary. 2. cancel_query never called from Java on coordinator cancel — the CancellationToken was registered in QUERY_REGISTRY but nothing fired it. Add CancellableExchangeSink SPI interface and implement cancel() on AbstractDatafusionReduceSink to call NativeBridge.cancelQuery(taskId). Wire it in LocalStageExecution.cancel() and failFromChild() so both the task-cancel and child-failure paths fire the token before close(), letting the drain unblock immediately instead of running to completion. 3. execute_indexed_with_context planning phase not cancellable — the entire substrait decode + plan + execute_stream sequence ran without checking the token. Split into a thin public wrapper that fetches the token and calls cancellable(), and an inner function that holds the original body. Signed-off-by: Aravind Sagar <sagarara@amazon.com>
…ensearch-project#21908) ArrowCalciteTypes.toArrow (used by the late-materialization stitch to build the fetch-stage output schema) had no case for SMALLINT/TINYINT, so a QTF query whose above-anchor fields include a short/byte column (e.g. clickbench Age under sort+head) threw "Unsupported Calcite type: SMALLINT" and the whole query 500'd. Map SMALLINT -> Int(16,true) and TINYINT -> Int(8,true), matching the wire Arrow types the data node emits (ShortParquetField / ByteParquetField) so the Stitcher's copyFromSafe sees aligned types. Adds ArrowCalciteTypesTests plus PplClickBenchIT.testQtfFetchOfShortAboveAnchorField proving a short above-anchor field round-trips through the QTF stitch end-to-end. Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…ry/BinaryView mismatch on reduce sink (opensearch-project#21910) Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
…project#21909) Recovers 8 currently-failing QA cases with test-resource-only changes (no production code): lookup_join_queries (6 tests, q1-q6): the dataset has two indices (sales_data, user_lookup) with suffixed files, and the queries join across both. The helper declared only "sales_data", so the provisioner took the single-index path and looked for a non-existent datasets/lookup_join_queries/mapping.json, failing the whole suite with "Resource not found". Declare both indices so the provisioner uses the per-index suffixed files, matching MultiIndexQueriesTestHelper. rex_command goldens (2 tests, q12/q15): the expected results were wrong. q12: rex on `message` captures reqid=abc-123 for all three RequestId docs; only the REPORT doc has Duration, so `where reqid is not null` yields 3 rows (two with null duration), not 1. q15: the rex regex's \w+ matches SELECT as well as HTTP methods, so docs with status>=500 form 5 groups (GET/500=2, PUT/500, GET/503, SELECT/500, POST/500), not the 2 the golden claimed. Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…ng viable backends (opensearch-project#21903) * Filter rule: respect nested scalar-function capabilities when computing viable backends Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> * remove @AwaitsFix from IT Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> * fixed mockdatafusionbackend capability Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> * skip Calcite-internal value constructors in filterrule Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> * Rename PredicateAnalysis to PredicateContents, add deeper-nesting test Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> * Updated tests and todos Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com> --------- Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
…ensearch-project#21906) Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…pensearch-project#21912) Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…21897) Co-authored-by: Craig Perkins <cwperx@amazon.com> Signed-off-by: Craig Perkins <cwperx@amazon.com>
…ct#21896) Signed-off-by: Andrew Ross <andrross@amazon.com>
…project#21916) Co-authored-by: Sandesh Kumar <sandeshkr419@gmail.com> Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…ilt-ins (opensearch-project#21823) Co-authored-by: Sandesh Kumar <sandeshkr419@gmail.com> Signed-off-by: Songkan Tang <songkant@amazon.com>
Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…TP-disconnect cancellation (opensearch-project#21917) Remove manual task register/unregister and rely on parent task in local transport call. Also update the explain endpoint to use our local transport call. Adds AnalyticsQueryTaskCleanupIT covering task cleanup on success and on cancel. Signed-off-by: Marc Handalian <marc.handalian@gmail.com> Co-authored-by: Sandesh Kumar <sandeshkr419@gmail.com>
…project#21920) Routes PPL percentile_approx(field, p) through the analytics-engine to DataFusion's builtin approx_percentile_cont as a single-stage gather-on- coordinator aggregate. PplAggregateCallRewriter strips the PPL type-flag arg and normalises SymbolFlag literals to VARCHAR (isthmus's LiteralConverter rejects unregistered enum classes). DataFusionFragmentConvertor's literal-arg normaliser rescales the percentile from PPL's [0, 100] to DataFusion's [0, 1] convention at substrait emission. OpenSearchAggregateRule skips backend-viability checks on null-FieldType metadata arg columns. Adds CoordinatorReduceIT.testPercentileApproxAcrossShards. Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>
…opensearch-project#21957) A shard whose parquet files have fewer columns than the widened union schema (built for alias / index-pattern queries by widen_schema_from_plan) could fail planning with "Cannot merge statistics with different number of columns". DataFusion's runtime-global file statistics cache is keyed by path + size + last_modified (not schema), so a Statistics computed during an earlier narrow read is returned for a later widened read; merging the cached narrow stats against freshly-computed widened stats blows up. Disable stat collection on the ListingTable when widening changed the schema. Non-widened (single-index) scans keep full stats. Added a TODO with better long-term options (schema-aware cache, normalize cached stats, upstream fix). Adds a Rust regression test (widened_scan_over_narrower_files_does_not_fail_ stats_merge) that reproduces the warm-cache straddle and fails without the fix. Also unmutes MultiIndexQueriesPplIT, which exercises the multi-index path: its @AwaitsFix was masking three stale test queries referencing renamed dataset fields (latency_ms -> latency, exception_type -> level), now fixed. All 14 queries pass. Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
… patterns, multi index queries (opensearch-project#21935) * add limit on max shards per query for alias, index patterns, multi index queries Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * spotless Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * test fixes Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * add null check for tests Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * update to use dynamic setting rather than static state lookup Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * spotless Signed-off-by: Marc Handalian <marc.handalian@gmail.com> --------- Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…search-project#21971) Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…roject#21716) Signed-off-by: gesong.samuel <gesong.samuel@bytedance.com> Co-authored-by: Andrew Ross <andrross@amazon.com>
…pensearch-project#21974) We could throw an exception in the KafkaPartitionConsumer constructor, and we were doing that after opening the Kafka Consumer (i.e. the thing that connects to Kafka). In that case, there was no good way to close the Consumer, resulting in a connection leak. This change applies the idea that "heavy" resource allocation should either come right at the end of a constructor, or should be moved into another method. In this case, I went with another method. Signed-off-by: Michael Froh <msfroh@apache.org>
…ct#21979) Signed-off-by: Varun Bharadwaj <varunbharadwaj1995@gmail.com>
…NT_DISTINCT (opensearch-project#21961 follow-up) (opensearch-project#21981) Distinctness is global and cannot be reduced additively across shards. Yesterday's opensearch-project#21961 fix routed every isDistinct aggregate through the single-stage gather path, which corrected over-counting but lost dc()'s approximate semantics and shipped raw rows. PPL dc / distinct_count are documented as approximate and should run as HLL. Resolve at the operator layer: * AggregateFunction.resolveOperator(...) — single source of truth that collapses exact single-arg COUNT(DISTINCT x) onto SqlStdOperatorTable.APPROX_COUNT_DISTINCT. * OpenSearchAggregateRule consults the SPI (no DC-specific code in the rule), rebuilding the call once when the operator changes. * OpenSearchAggregateSplitRule.shouldSkipPartialFinalSplit now skips for Type.APPROXIMATE alongside the existing STATE_EXPANDING and residual-isDistinct cases — the rewritten aggregate runs once at the coordinator over gathered rows. End-to-end: dc()/distinct_count() return the HLL estimate (correct, no over-count); multi-arg COUNT(DISTINCT a,b) still falls through to the safe single-stage path. Distributed sketch-merge (per-shard 16KB sketches over the wire) is a follow-up that needs SETUP_PARTIAL_AGGREGATE wiring + state-suffix column plumbing through derive_schema_from_partial_plan. Tests: 5 unit cases for resolveOperator; AggregateRuleTests verifies the planner emits APPROX_COUNT_DISTINCT for COUNT(DISTINCT). TwoShardAggregationIT (26/26 incl. previously-xfailed dc_label / distinct_count_label / distinct_count_by_cat) and CoordinatorReduceIT (20/20) green; spotless clean. Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>
…ect#21976) * Improve sandbox check and add Cargo.lock for caching Signed-off-by: Peter Zhu <zhujiaxi@amazon.com> * Remove restore keys Signed-off-by: Peter Zhu <zhujiaxi@amazon.com> --------- Signed-off-by: Peter Zhu <zhujiaxi@amazon.com>
…rage (opensearch-project#21982) Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…ject#21978) Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
* Fix temp lucene_gen folder deletion Signed-off-by: rayshrey <rayshrey@amazon.com> * Review fixes Signed-off-by: rayshrey <rayshrey@amazon.com> --------- Signed-off-by: rayshrey <rayshrey@amazon.com>
…earch-project#21988) The scan path already widens each shard's schema to the plan's base_schema, but the indexed-executor path rebuilds its own schema from build_segments (local shard only) and skipped widening. Fix: widen to base_schema there too, so absent columns are appended as nullable and null-filled at read time. Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
JEP 467 has introduced a tripe-slash syntax for markdown javadoc content. A few places were using `////////` style dividers that was confusing spotless, so I've reformated them to use a different divider syntax. Signed-off-by: Andrew Ross <andrross@amazon.com> Co-authored-by: Peter Zhu <zhujiaxi@amazon.com>
…-project#21904) * Add bloom filter pruning on the indexed parquet read path Integrate per-row-group bloom filter checks into the indexed execution prefetch phase. Reuses DataFusion's PruningPredicate::prune() with a BloomFilterStatistics PruningStatistics impl — same pattern DataFusion uses internally for its vanilla ParquetExec bloom filter pruning. Execution order in prefetch_rg: 1. Page pruning (in-memory min/max stats — free) 2. Early exit if all pages pruned 3. Bloom filter check (small IO read ~200µs via object store) 4. FFM collector call (Lucene round-trip ~1-5ms) When enabled, literal_columns() identifies which columns to check, bloom filter bytes are read from the object store, and PruningPredicate::prune() evaluates whether the RG can be skipped. Java: - datafusion.indexed.bloom_filter_on_read (Dynamic, NodeScope, default true) - WireConfigSnapshot: new field at offset 72 (i32, 0/1) Rust: - New: indexed_table/bloom_pruner.rs (BloomFilterStatistics + bloom_prune_rg) - SingleCollectorEvaluator: BloomConfig struct, bloom check after page pruning - DatafusionQueryConfig: bloom_filter_on_read in wire format * Wire bloom filter pruning metrics Add rg_bloom_pruned (Count) and bloom_filter_eval_time (Time) to PartitionMetrics and StreamMetrics. Recorded in BloomConfig and incremented in prefetch_rg when bloom filter prunes a row group. Signed-off-by: Bukhtawar Khan <bukhtawa@amazon.com>
…opensearch-project#21987) * Added tests for Filter Delegation to test failures Signed-off-by: expani <anijainc@amazon.com> * Using explain instead of 2 API calls Signed-off-by: expani <anijainc@amazon.com> * Added both profile and non-profile assertions for each shape Signed-off-by: expani <anijainc@amazon.com> * Fix for ClassCast exception in during combining predicates Signed-off-by: expani <anijainc@amazon.com> --------- Signed-off-by: expani <anijainc@amazon.com>
…ject#21969) Signed-off-by: Andrew Ross <andrross@amazon.com>
…#21994) * Add fix for the table_provider Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> * Add witness test for sequential chunk execution within a partition Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> * Pass bloom_config to SingleCollectorEvaluator in witness test Aligns the witness test's call site with the 12-arg signature added by the bloom-filter pruning change on main. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> * Retrigger CI Unrelated flaky test on previous run: LuceneDeleteExecutionEngineTests.testConcurrentGenerationRotation (analytics-backend-lucene plugin; this PR is Rust-only in analytics-backend-datafusion/rust/src/indexed_table). Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> * Apply spotless formatting fix spotless 8.6.0 (landed on main via opensearch-project#21977) reformats inline-comment spacing in COMMAS enum constant. Result of running `./gradlew spotlessApply` on analytics-backend-datafusion. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> --------- Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
…earch-project#21990) Delegated filter scans each built their own IndexSearcher over the shard's DirectoryReader. When two scans over the same reader share the node query cache (e.g. a self-union from multisearch/append), a Weight cached against one searcher's reader-context is evaluated via the other's, tripping Lucene's "top-reader used to create Weight is not the same as the current reader's top-reader" assertion — fatal across the FFM upcall, crashing the data node. LuceneReader now exposes one lazily-built, shared IndexSearcher per reader; the delegation and count-scan paths use it instead of constructing their own. The shared cache stays attached, so cross-scan cache reuse is preserved. Tests: shared-searcher reuse across two scans with a shared cache, and concurrent scorers across partitions (multi-partition thread-safety). Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…search-project#22004) Signed-off-by: Andriy Redko <drreta@gmail.com>
opensearch-project#22005) * Fix TopK oversampling to use the user's limit, not the system size cap A query like `... | stats count() by x | sort - c | head 10` was making each shard fetch way too many rows. The SQL plugin wraps every query in a system size-limit (default 10000), and the oversampling code was reading that 10000 instead of the user's `head 10` — so shards fetched ~30000 rows each instead of ~30. The fix: when there are two stacked sorts (the system cap over the user's head), use the inner one (the real limit) to size the per-shard fetch. Signed-off-by: Marc Handalian <marc.handalian@gmail.com> * spotless Signed-off-by: Marc Handalian <marc.handalian@gmail.com> --------- Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…IME_PLUS on DataFusion (opensearch-project#21991) * [analytics-engine] Support PPL date_add/date_sub by lowering to DATETIME_PLUS on DataFusion Signed-off-by: Lantao Jin <ltjin@amazon.com> * make javadoc concise Signed-off-by: Lantao Jin <ltjin@amazon.com> --------- Signed-off-by: Lantao Jin <ltjin@amazon.com>
* Add datafusion.spill_directory setting Adds one new setting in sandbox/plugins/analytics-backend-datafusion: - datafusion.spill_directory (NodeScope, Final). Empty default preserves current behaviour for self-managed clusters that never configured this setting -- the spill directory still resolves to path.data[0]/../tmp. AOS sets this to the spill volume mount point via opensearch.yml. Validator does a syntactic parse of the path; existence is intentionally not checked because the directory may be created later by a host-fleet boot script (first-boot mount), and runtime spill writes will surface any permission issues at first spill with a clear DataFusion error. resolveSpillDirectory() picks the configured value when set, falls back to the legacy path.data[0]/../tmp default otherwise. The fallback preserves backward compatibility. Signed-off-by: Gaurav Singh <snghsvn@amazon.com>
…r native memory tracking (opensearch-project#21973) * Thread context_id through fetch-by-row-ids and indexed query paths for native memory tracking wrap_stream_as_handle() and execute_indexed_query() hardcoded context_id=0 when creating QueryTrackingContext, making queries through these paths invisible to Search Backpressure's NativeMemoryUsageTracker. SBP could not monitor or cancel these tasks based on native memory pressure. Fix: thread the task's context_id from the Java layer through FFM into both Rust functions so queries are properly registered in QUERY_REGISTRY. Signed-off-by: Himshikha Gupta <himshikh@amazon.com>
…ch-project#21825) * Adding reload for SSE enabled blob store during repo metadata update Signed-off-by: Raghuvansh Raj <raghraaj@amazon.com> (cherry picked from commit d109033) * Added UTs Signed-off-by: Kamal Nayan <askkamal@amazon.com> * Added lucene temp file as excluded subdirectories list Signed-off-by: Kamal Nayan <askkamal@amazon.com> --------- Signed-off-by: Raghuvansh Raj <raghraaj@amazon.com> Signed-off-by: Kamal Nayan <askkamal@amazon.com> Co-authored-by: Raghuvansh Raj <raghraaj@amazon.com> Co-authored-by: Kamal Nayan <askkamal@amazon.com>
…opensearch-project#21877) Fans out the analytics stats rollup to every cluster node and renders one entry per node, mirroring _nodes/stats's shape: { "_nodes": { "total": N, "successful": N, "failed": 0 }, "cluster_name": "...", "nodes": { "<node-id>": { "analytics": { "queries": {...}, "stages_by_type": {...}, "fragments": {...} } }, ... } } PR opensearch-project#21796 only returned the rollup for whichever node the REST client happened to land on. With the request load-balancer round-robining, a single GET only saw a slice of the cluster's activity. Promoting the endpoint to a TransportNodesAction makes it cluster-wide so existing tooling can poll one URL and see every node's distribution. Each node still computes its own percentiles locally — the coordinator collects per-node AnalyticsStats and renders them side-by-side. No histogram merging is required because there is no top-level rollup; if one is added later it will need the histograms on the wire. Changes: - AnalyticsStats and its nested records implement Writeable. - New transport package with AnalyticsStatsAction, AnalyticsStatsRequest, AnalyticsStatsNodeRequest, AnalyticsStatsNodeResponse, AnalyticsStatsResponse, and TransportAnalyticsStatsAction. - RestAnalyticsStatsAction dispatches via NodesResponseRestListener, dropping its direct AnalyticsStatsCollector reference. - AnalyticsPlugin registers the action in getActions() and constructs the REST handler without the collector. - AnalyticsStatsApiIT updated to assert the cluster-wide shape and to verify at least one node recorded a query, stage, and fragment. - AnalyticsStatsTests covers the StreamInput/StreamOutput round-trip for every record. Signed-off-by: Michael Oviedo <mikeovi@amazon.com>
…rrelevant subsearch limit (opensearch-project#22012) Signed-off-by: Lantao Jin <ltjin@amazon.com>
…arch-project#22003) Surfaces Arrow Flight's per-frame metadata via ArrowBatchResponse so actions can attach opaque bytes to a batch and read them on the consumer without any transport-SPI changes. Wire path: putNext(ArrowBuf) on send, getLatestMetadata() on receive (both already in Arrow Flight 18.x). The transport copies bytes off the wire on the receive side so the byte[] outlives the stream cursor. API surface: - ArrowBatchResponse(VectorSchemaRoot, byte[] metadata) — send-side ctor. - ArrowBatchResponse#getMetadata() — receive-side accessor. - ArrowStreamInput#getMetadata() — default null; NativeArrow input carries it through. Transport plumbing: - FlightServerChannel.sendBatch(header, output, byte[]) — when non-null, allocates an ArrowBuf from the channel allocator and uses putNext(buf). - FlightOutboundHandler.processBatchTask reads ArrowBatchResponse.getMetadata() and threads it through. - FlightTransportResponse.nextResponse pulls flightStream.getLatestMetadata() per frame, copies into byte[]. No changes to public SPI in server/. No FlightTransportChannel surface changes. Native Arrow path only — byte-serialized path unchanged. Tests: - StreamMetadataIT — two cases (profile=true: metadata observed once on last batch with bytes intact; profile=false: zero metadata observations). - FlightOutboundHandlerTests updated for new 3-arg sendBatch signature. - NativeArrowTransportIT + all unit tests still pass. Docs: native-arrow-transport-design.md gains an "Application Metadata" section; server-side-streaming-guide.md cross-links to it. Signed-off-by: Rishabh Maurya <rishabhmaurya05@gmail.com>
Extends the profile API to return per-shard DataFusion execution metrics (output_rows, elapsed_compute, scan_time, row_groups_pruned, etc.) in the profile output as 'data_node_metrics' per task. Uses the ArrowBatchResponse metadata channel (merged in opensearch-project#22003) to transmit metrics in-band on the last data batch. No sentinel frames, no transport SPI changes, no threading races. Data node side: - AnalyticsSearchService extracts metrics after stream exhaustion - channelResponseHandler buffers one batch ahead; attaches metrics to the final batch via FragmentExecutionArrowResponse(root, metadata) Coordinator side: - handleStreamResponse reads last.getMetadata() after the stream loop - StreamingResponseListener.onStreamComplete(bytes) passes to task - ShardFragmentStageExecution stores on StageTask.setDataNodeMetrics - QueryProfileBuilder parses JSON into TaskProfile.dataNodeMetrics Also includes: - Profile flag propagation (QueryContext → FragmentExecutionRequest) - Rust FFM: df_stream_get_metrics extracts ExecutionPlan.metrics() - Debug logging of metrics at shard level (guarded by isDebugEnabled) - Integration test: ExplainApiIT.testExplainTasksHaveDataNodeMetrics Signed-off-by: Finn Carroll <carrofin@amazon.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
20 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
[Describe what this change achieves]
Related Issues
Resolves #[Issue number to be closed when this PR is merged]
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.