comphead commented on code in PR #26067:
URL: https://github.com/apache/datafusion/pull/26067#discussion_r4351060606
##########
datafusion/sqllogictest/test_files/window.slt:
##########
@@ -7207,23 +7206,29 @@ select count(*) over (order by l) from (values ([1]),
([2])) t(l)
1
2
-# but a list whose elements are nested (struct, list) plans and then fails at
-# execution: the peer comparison in the RANGE frame goes through
-# `ScalarValue::partial_cmp`, which cannot compare nested list elements, while
-# the sort that produced the input uses `make_comparator` and can. Documented
-# here so the fix to the scalar comparison can flip these expectations.
-query error DataFusion error: Internal error: Uncomparable values:
List\(\[\{c0: 1\}\]\), List\(\[\{c0: 1\}\]\)
+# NULL elements must form the same peers as Arrow's list sort order.
+query I
+select count(*) over (order by l nulls first) from (values ([NULL::int]),
([1]), ([NULL::int])) t(l)
+----
+2
+2
+3
+
+# Lists containing structs or lists also form RANGE peer groups.
+query I
select count(*) over (order by l) from (values ([struct(1)]), ([struct(2)]))
t(l)
+----
+1
+2
-query error DataFusion error: Internal error: Uncomparable values:
List\(\[\[1\]\]\), List\(\[\[1\]\]\)
+query I
select count(*) over (order by l) from (values ([[1]]), ([[2]])) t(l)
Review Comment:
Both nested queries here use distinct keys, so they only show that the
`Uncomparable values` error is gone. They do not exercise peers, which the
comment above them claims. Adding a tied row, for example `([[1]]), ([[1]]),
([[2]])` expecting `2, 2, 3`, and the same for `[struct(1)]`, would cover it.
##########
datafusion/core/tests/execution/window_nested_range.rs:
##########
@@ -0,0 +1,321 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Nested RANGE peers through public physical execution APIs.
+//! Construct plans directly to test peer boundaries independently of SQL type
admission.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use arrow::array::{ArrayRef, AsArray, Int32Array, Int64Array, ListArray,
StructArray};
+use arrow::buffer::{NullBuffer, OffsetBuffer};
+use arrow::compute::concat_batches;
+use arrow::datatypes::{DataType, Field, Int64Type, SchemaRef};
+use arrow::record_batch::RecordBatch;
+use datafusion::datasource::memory::MemorySourceConfig;
+use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::TaskContext;
+use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
+use datafusion::physical_plan::streaming::{PartitionStream,
StreamingTableExec};
+use datafusion::physical_plan::windows::{
+ BoundedWindowAggExec, WindowAggExec, create_window_expr,
+};
+use datafusion::physical_plan::{
+ ExecutionPlan, ExecutionPlanProperties, InputOrderMode,
SendableRecordBatchStream,
+ collect,
+};
+use datafusion::prelude::SessionContext;
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ WindowFrame, WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition,
+};
+use datafusion_functions_aggregate::sum::sum_udaf;
+use datafusion_physical_expr::PhysicalSortExpr;
+use datafusion_physical_expr::expressions::col;
+use datafusion_physical_expr_common::sort_expr::LexOrdering;
+use futures::{FutureExt, StreamExt};
+
+fn nested_batches() -> Result<Vec<RecordBatch>> {
+ // Three nested keys: a NULL container, a nested NULL, and nested 1.
+ // Hidden values under the two NULL containers deliberately differ.
+ let values = [Some(10), Some(99), None, None, None, Some(1)];
+ let values: ArrayRef = Arc::new(Int32Array::from(values.to_vec()));
+ let lists: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("item", DataType::Int32, true)),
+ OffsetBuffer::from_lengths([1; 6]),
+ Arc::clone(&values),
+ None,
+ ));
+ let structs: ArrayRef = Arc::new(StructArray::new(
+ vec![Arc::new(Field::new("item", DataType::Int32, true))].into(),
+ vec![values],
+ None,
+ ));
+ let nulls = NullBuffer::from(vec![false, false, true, true, true, true]);
+ let list_field = Arc::new(Field::new("item", lists.data_type().clone(),
true));
+ let keys: [ArrayRef; 3] = [
+ Arc::new(ListArray::new(
+ Arc::clone(&list_field),
+ OffsetBuffer::from_lengths([1; 6]),
+ Arc::clone(&lists),
+ Some(nulls.clone()),
+ )),
+ Arc::new(ListArray::new(
+ Arc::new(Field::new("item", structs.data_type().clone(), true)),
+ OffsetBuffer::from_lengths([1; 6]),
+ structs,
+ Some(nulls.clone()),
+ )),
+ Arc::new(StructArray::new(
+ vec![list_field].into(),
+ vec![lists],
+ Some(nulls),
+ )),
+ ];
+ keys.into_iter()
+ .map(|keys| {
+ // The tie separates the third nested NULL from the preceding
peers.
+ let tie: ArrayRef = Arc::new(Int32Array::from(vec![0, 0, 0, 0, 1,
0]));
+ Ok(RecordBatch::try_from_iter(vec![
+ ("key", keys),
+ ("tie", tie),
+ // In Linear mode, the final row advances another partition's
peers.
+ (
+ "p",
+ Arc::new(Int32Array::from(vec![0, 0, 0, 0, 0, 1])) as
ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int64Array::from(vec![1, 2, 4, 8, 16, 32])) as
ArrayRef,
+ ),
+ ])?)
+ })
+ .collect()
+}
+
+#[tokio::test]
+async fn nested_range_current_row_physical_operators() -> Result<()> {
+ use WindowFrameBound::{CurrentRow, Following, Preceding};
+
+ for batch in nested_batches()? {
+ let schema = batch.schema();
+ let order_by = vec![
+ PhysicalSortExpr::new_default(col("key", &schema)?),
+ PhysicalSortExpr::new_default(col("tie", &schema)?),
+ ];
+ // Repeated peers cross input batch boundaries.
+ let batches = (0..6).map(|row| batch.slice(row, 1)).collect();
+ let source = MemorySourceConfig::try_new(&[batches],
Arc::clone(&schema), None)?
+ .try_with_sort_information(vec![
+ LexOrdering::new(order_by.clone()).unwrap(),
+ ])?;
+ let input: Arc<dyn ExecutionPlan> =
DataSourceExec::from_data_source(source);
+ for (start, end, expected) in [
+ (
+ Preceding(ScalarValue::UInt64(None)),
+ CurrentRow,
+ [3, 3, 15, 15, 31, 63],
+ ),
+ (CurrentRow, CurrentRow, [3, 3, 12, 12, 16, 32]),
+ (
+ CurrentRow,
+ Following(ScalarValue::UInt64(None)),
+ [63, 63, 60, 60, 48, 32],
+ ),
+ ] {
+ for mode in [
+ None,
+ Some(InputOrderMode::Sorted),
+ Some(InputOrderMode::Linear),
+ ] {
+ // The planner uses WindowAggExec for UNBOUNDED FOLLOWING.
+ if mode.is_some() && end.is_unbounded() {
+ continue;
+ }
+ let partition_by = if mode == Some(InputOrderMode::Linear) {
+ vec![col("p", &schema)?]
+ } else {
+ vec![]
+ };
+ let expr = create_window_expr(
Review Comment:
The plan setup (`create_window_expr`, the exec construction and the
`ordering_satisfy_requirement` assertion) is repeated almost verbatim in
`nested_range_emits_completed_peers_before_eof`. A small helper taking the mode
and frame would remove about 40 lines.
Also, the `WindowAggExec` and `Sorted` cases for the `List<List>` and
`List<Struct>` keys are reachable from SQL, since a list is accepted as the
first ORDER BY key (`ORDER BY l, tie`). They would fit `window.slt` with `set
datafusion.execution.batch_size = 1` to get the batch-boundary crossing. The
struct key, `Linear` mode and the pending-stream case need to stay in Rust.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
