You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "alamb (via GitHub)" <gi...@apache.org> on 2023/08/11 15:19:36 UTC
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #7250: [NOT READY FOR REVIEW] Native TopK Operator
alamb commented on code in PR #7250:
URL: https://github.com/apache/arrow-datafusion/pull/7250#discussion_r1291449810
##########
datafusion/core/tests/sqllogictests/test_files/window.slt:
##########
@@ -3208,8 +3209,8 @@ SELECT
ORDER BY C3
LIMIT 5
----
-0.970671228336 0.970671228336
Review Comment:
these have the same problem -- that the top 5 values have the same value for the `c3` column
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -847,29 +853,54 @@ impl ExecutionPlan for SortExec {
trace!("End SortExec's input.execute for partition: {}", partition);
- let mut sorter = ExternalSorter::new(
- partition,
- input.schema(),
- self.expr.clone(),
- context.session_config().batch_size(),
- self.fetch,
- execution_options.sort_spill_reservation_bytes,
- execution_options.sort_in_place_threshold_bytes,
- &self.metrics_set,
- context.runtime_env(),
- );
+ if let Some(fetch) = self.fetch.as_ref() {
Review Comment:
In this prototype I simply used the `TopK` implementation for `SortExec` when there was a fetch as it produces the same output.
However, longer term I think we should make a real `TopKExec` `ExecutionPlan` so that the optimizers know about it and can avoid repartitioning / trying to do anything else fancy
##########
datafusion/core/src/physical_plan/topk/mod.rs:
##########
@@ -0,0 +1,516 @@
+// 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.
+
+//! TopK: Combination of Sort / LIMIT
+
+use arrow::{
+ compute::interleave,
+ row::{OwnedRow, RowConverter, Rows, SortField},
+};
+use std::{cmp::Ordering, sync::Arc};
+
+use arrow_array::{Array, ArrayRef, RecordBatch};
+use arrow_schema::SchemaRef;
+use datafusion_common::Result;
+use datafusion_execution::{
+ memory_pool::{MemoryConsumer, MemoryReservation},
+ runtime_env::RuntimeEnv,
+};
+use datafusion_physical_expr::PhysicalSortExpr;
+use hashbrown::HashMap;
+
+use crate::physical_plan::{stream::RecordBatchStreamAdapter, SendableRecordBatchStream};
+
+use super::metrics::{BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder};
+
+/// Global TopK
+///
+/// # Background
+///
+/// "Top K" is a common query optimization used for queries such as
+/// "find the top 3 customers by revenue". The (simplified) SQL for
+/// such a query might be:
+///
+/// ```sql
+/// SELECT customer_id, revenue FROM 'sales.csv' ORDER BY revenue DESC limit 3;
+/// ```
+///
+/// The simple plan would be:
+///
+/// ```
+/// > explain SELECT customer_id, revenue FROM sales ORDER BY revenue DESC limit 3;
+/// +--------------+----------------------------------------+
+/// | plan_type | plan |
+/// +--------------+----------------------------------------+
+/// | logical_plan | Limit: 3 |
+/// | | Sort: revenue DESC NULLS FIRST |
+/// | | Projection: customer_id, revenue |
+/// | | TableScan: sales |
+/// +--------------+----------------------------------------+
+/// ```
+///
+/// While this plan produces the correct answer, it will fully sorts the
+/// input before discarding everything other than the top 3 elements.
+///
+/// The same answer can be produced by simply keeping track of the top
+/// N elements, reducing the total amount of required buffer memory.
+///
+/// # Structure
+///
+/// This operator tracks the top K items using a `TopKHeap`.
+pub struct TopK {
+ /// schema of the output (and the input)
+ schema: SchemaRef,
+ /// Runtime metrics
+ metrics: TopKMetrics,
+ /// Reservation
+ reservation: MemoryReservation,
+ /// The target number of rows for output batches
+ batch_size: usize,
+ /// sort expressions
+ expr: Arc<[PhysicalSortExpr]>,
+ /// row converter, for sort keys
+ row_converter: RowConverter,
+ /// scratch space for converting rows
+ scratch_rows: Rows,
+ /// stores the top k values and their sort key values, in order
+ heap: TopKHeap,
+}
+
+impl TopK {
+ /// Create a new [`TopK`] that stores the top `k` values, as
+ /// defined by the sort expressions in `expr`.
+ // TOOD: make a builder or some other nicer API to avoid the
+ // clippy warning
+ #[allow(clippy::too_many_arguments)]
+ pub fn try_new(
+ partition_id: usize,
+ schema: SchemaRef,
+ expr: Vec<PhysicalSortExpr>,
+ k: usize,
+ batch_size: usize,
+ runtime: Arc<RuntimeEnv>,
+ metrics: &ExecutionPlanMetricsSet,
+ partition: usize,
+ ) -> Result<Self> {
+ let reservation = MemoryConsumer::new(format!("TopK[{partition_id}]"))
+ .register(&runtime.memory_pool);
+
+ let expr: Arc<[PhysicalSortExpr]> = expr.into();
+
+ let sort_fields: Vec<_> = expr
+ .iter()
+ .map(|e| {
+ Ok(SortField::new_with_options(
+ e.expr.data_type(&schema)?,
+ e.options,
+ ))
+ })
+ .collect::<Result<_>>()?;
+
+ let row_converter = RowConverter::new(sort_fields)?;
+ let scratch_rows = row_converter.empty_rows(
+ batch_size,
+ 20 * batch_size, // guestimate 20 bytes per row
+ );
+
+ Ok(Self {
+ schema,
+ metrics: TopKMetrics::new(metrics, partition),
+ reservation,
+ batch_size,
+ expr,
+ row_converter,
+ scratch_rows,
+ heap: TopKHeap::new(k),
+ })
+ }
+
+ /// Insert `batch`, remembering it if any of its values are among
+ /// the top k seen so far.
+ pub fn insert_batch(&mut self, batch: RecordBatch) -> Result<()> {
+ // Updates on drop
+ let _timer = self.metrics.baseline.elapsed_compute().timer();
+
+ let sort_keys: Vec<ArrayRef> = self
+ .expr
+ .iter()
+ .map(|expr| {
+ let value = expr.expr.evaluate(&batch)?;
+ Ok(value.into_array(batch.num_rows()))
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // reuse existing `Rows` to avoid reallocations
+ let rows = &mut self.scratch_rows;
+ rows.clear();
+ self.row_converter.append(rows, &sort_keys)?;
+
+ // TODO make this algorithmically better?:
+ // 1. only check topk values in rows
+ // 2. only do one update through top_k
+
+ let mut batch_entry = self.heap.register_batch(batch);
+ for (index, row) in rows.iter().enumerate() {
+ match self.heap.k_largest() {
+ // heap has k items, and the current row is not
+ // smaller than the curret smallest k value, skip
+ Some(largest) if largest.row.row() <= row => {}
+ // don't yet have k items or new item is greater than
+ // current min top k
+ None | Some(_) => {
+ self.heap.add(&mut batch_entry, row.owned(), index);
+ self.metrics.row_replacements.add(1);
+ }
+ }
+ }
+ self.heap.insert_batch_entry(batch_entry);
+
+ // update memory reservation
+ self.reservation.try_resize(self.size())?;
+ Ok(())
+ }
+
+ /// Returns the top k results broken into `batch_size` [`RecordBatch`]es
+ pub fn emit(self) -> Result<SendableRecordBatchStream> {
+ let Self {
+ schema,
+ metrics,
+ reservation: _,
+ batch_size,
+ expr: _,
+ row_converter: _,
+ scratch_rows: _,
+ heap,
+ } = self;
+ let _timer = metrics.baseline.elapsed_compute().timer(); // time updated on drop
+
+ let mut batch = heap.emit(schema.clone())?;
+ metrics.baseline.output_rows().add(batch.num_rows());
+
+ // break into record batches as needed
+ let mut batches = vec![];
+ loop {
+ if batch.num_rows() < batch_size {
+ batches.push(Ok(batch));
+ break;
+ } else {
+ batches.push(Ok(batch.slice(0, batch_size)));
+ batch = batch.slice(batch_size, batch.num_rows());
+ }
+ }
+ Ok(Box::pin(RecordBatchStreamAdapter::new(
+ schema,
+ futures::stream::iter(batches),
+ )))
+ }
+
+ /// return the size of memory used by this operator, in bytes
+ fn size(&self) -> usize {
+ std::mem::size_of::<Self>()
+ + self.row_converter.size()
+ + self.scratch_rows.size()
+ + self.heap.size()
+ }
+}
+
+struct TopKMetrics {
+ /// metrics
+ pub baseline: BaselineMetrics,
+
+ /// count of how many rows were replaced in the heap
+ pub row_replacements: Count,
+}
+
+impl TopKMetrics {
+ fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
+ Self {
+ baseline: BaselineMetrics::new(metrics, partition),
+ row_replacements: MetricBuilder::new(metrics)
+ .counter("row_replacements", partition),
+ }
+ }
+}
+
+/// This structure keeps at most the *smallest* k items, using the
+/// [arrow::row] format for sort keys. While it is called "topK" for
+/// values like `1, 2, 3, 4, 5` the "top 3" really means the
+/// *smallest* 3 , `1, 2, 3`, not the *largest* 3 `3, 4, 5`.
+///
+/// Using the `Row` format handles things such as ascending vs
+/// descending and nulls first vs nulls last.
+///
+/// It doesn't use `BinaryHeap` in the Rust standard library because
+/// it is important to check the current minimum value in the heap
+/// prior to creating a new value to insert.
+struct TopKHeap {
+ /// The maximum size of this heap.
+ k: usize,
+ /// Storage for up at most `k` items, in ascending
+ /// order. `inner[0]` holds the smallest value of the smallest k
+ /// so far, `inner[len-1]` holds the largest value smallest k so far.
+ inner: Vec<TopKRow>,
+ /// Storage the original row values (TopKRow only has the sort key)
+ store: RecordBatchStore,
+ /// The size of all `OwnedRows`s held by this heap
+ owned_row_bytes: usize,
+}
+
+impl TopKHeap {
+ fn new(k: usize) -> Self {
+ assert!(k > 0);
+ Self {
+ k,
+ inner: Vec::with_capacity(k),
+ store: RecordBatchStore::new(),
+ owned_row_bytes: 0,
+ }
+ }
+
+ /// Register a [`RecordBatch`] with the heap, returning the
+ /// appropriate entry
+ pub fn register_batch(&mut self, batch: RecordBatch) -> RecordBatchEntry {
+ self.store.register(batch)
+ }
+
+ /// Insert a [`RecordBatchEntry`] created by a previous call to
+ /// [`Self::register_batch`] into storage.
+ pub fn insert_batch_entry(&mut self, entry: RecordBatchEntry) {
+ self.store.insert(entry)
+ }
+
+ /// Returns the largest value stored by the heap if there are k
+ /// items, otherwise returns None
+ fn k_largest(&self) -> Option<&TopKRow> {
+ if self.inner.len() < self.k {
+ None
+ } else {
+ self.inner.last()
+ }
+ }
+
+ /// Adds `row` to this heap. If inserting this new item would
+ /// increase the size past `k`, removes the previously smallest
+ /// item.
+ fn add(&mut self, batch_entry: &mut RecordBatchEntry, row: OwnedRow, index: usize) {
+ assert!(self.inner.len() <= self.k);
+
+ batch_entry.uses += 1;
+
+ self.owned_row_bytes += owned_row_size(&row);
+
+ // put the new row into the correct location to maintain that
+ // self.inner is sorted in descending order
+ let insertion_point = self
Review Comment:
here is the other algorithmic thing
##########
datafusion/core/src/physical_plan/topk/mod.rs:
##########
@@ -0,0 +1,516 @@
+// 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.
+
+//! TopK: Combination of Sort / LIMIT
+
+use arrow::{
+ compute::interleave,
+ row::{OwnedRow, RowConverter, Rows, SortField},
+};
+use std::{cmp::Ordering, sync::Arc};
+
+use arrow_array::{Array, ArrayRef, RecordBatch};
+use arrow_schema::SchemaRef;
+use datafusion_common::Result;
+use datafusion_execution::{
+ memory_pool::{MemoryConsumer, MemoryReservation},
+ runtime_env::RuntimeEnv,
+};
+use datafusion_physical_expr::PhysicalSortExpr;
+use hashbrown::HashMap;
+
+use crate::physical_plan::{stream::RecordBatchStreamAdapter, SendableRecordBatchStream};
+
+use super::metrics::{BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder};
+
+/// Global TopK
+///
+/// # Background
+///
+/// "Top K" is a common query optimization used for queries such as
+/// "find the top 3 customers by revenue". The (simplified) SQL for
+/// such a query might be:
+///
+/// ```sql
+/// SELECT customer_id, revenue FROM 'sales.csv' ORDER BY revenue DESC limit 3;
+/// ```
+///
+/// The simple plan would be:
+///
+/// ```
+/// > explain SELECT customer_id, revenue FROM sales ORDER BY revenue DESC limit 3;
+/// +--------------+----------------------------------------+
+/// | plan_type | plan |
+/// +--------------+----------------------------------------+
+/// | logical_plan | Limit: 3 |
+/// | | Sort: revenue DESC NULLS FIRST |
+/// | | Projection: customer_id, revenue |
+/// | | TableScan: sales |
+/// +--------------+----------------------------------------+
+/// ```
+///
+/// While this plan produces the correct answer, it will fully sorts the
+/// input before discarding everything other than the top 3 elements.
+///
+/// The same answer can be produced by simply keeping track of the top
+/// N elements, reducing the total amount of required buffer memory.
+///
+/// # Structure
+///
+/// This operator tracks the top K items using a `TopKHeap`.
+pub struct TopK {
+ /// schema of the output (and the input)
+ schema: SchemaRef,
+ /// Runtime metrics
+ metrics: TopKMetrics,
+ /// Reservation
+ reservation: MemoryReservation,
+ /// The target number of rows for output batches
+ batch_size: usize,
+ /// sort expressions
+ expr: Arc<[PhysicalSortExpr]>,
+ /// row converter, for sort keys
+ row_converter: RowConverter,
+ /// scratch space for converting rows
+ scratch_rows: Rows,
+ /// stores the top k values and their sort key values, in order
+ heap: TopKHeap,
+}
+
+impl TopK {
+ /// Create a new [`TopK`] that stores the top `k` values, as
+ /// defined by the sort expressions in `expr`.
+ // TOOD: make a builder or some other nicer API to avoid the
+ // clippy warning
+ #[allow(clippy::too_many_arguments)]
+ pub fn try_new(
+ partition_id: usize,
+ schema: SchemaRef,
+ expr: Vec<PhysicalSortExpr>,
+ k: usize,
+ batch_size: usize,
+ runtime: Arc<RuntimeEnv>,
+ metrics: &ExecutionPlanMetricsSet,
+ partition: usize,
+ ) -> Result<Self> {
+ let reservation = MemoryConsumer::new(format!("TopK[{partition_id}]"))
+ .register(&runtime.memory_pool);
+
+ let expr: Arc<[PhysicalSortExpr]> = expr.into();
+
+ let sort_fields: Vec<_> = expr
+ .iter()
+ .map(|e| {
+ Ok(SortField::new_with_options(
+ e.expr.data_type(&schema)?,
+ e.options,
+ ))
+ })
+ .collect::<Result<_>>()?;
+
+ let row_converter = RowConverter::new(sort_fields)?;
+ let scratch_rows = row_converter.empty_rows(
+ batch_size,
+ 20 * batch_size, // guestimate 20 bytes per row
+ );
+
+ Ok(Self {
+ schema,
+ metrics: TopKMetrics::new(metrics, partition),
+ reservation,
+ batch_size,
+ expr,
+ row_converter,
+ scratch_rows,
+ heap: TopKHeap::new(k),
+ })
+ }
+
+ /// Insert `batch`, remembering it if any of its values are among
+ /// the top k seen so far.
+ pub fn insert_batch(&mut self, batch: RecordBatch) -> Result<()> {
+ // Updates on drop
+ let _timer = self.metrics.baseline.elapsed_compute().timer();
+
+ let sort_keys: Vec<ArrayRef> = self
+ .expr
+ .iter()
+ .map(|expr| {
+ let value = expr.expr.evaluate(&batch)?;
+ Ok(value.into_array(batch.num_rows()))
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // reuse existing `Rows` to avoid reallocations
+ let rows = &mut self.scratch_rows;
+ rows.clear();
+ self.row_converter.append(rows, &sort_keys)?;
+
+ // TODO make this algorithmically better?:
+ // 1. only check topk values in rows
+ // 2. only do one update through top_k
+
+ let mut batch_entry = self.heap.register_batch(batch);
+ for (index, row) in rows.iter().enumerate() {
+ match self.heap.k_largest() {
+ // heap has k items, and the current row is not
+ // smaller than the curret smallest k value, skip
+ Some(largest) if largest.row.row() <= row => {}
+ // don't yet have k items or new item is greater than
+ // current min top k
+ None | Some(_) => {
+ self.heap.add(&mut batch_entry, row.owned(), index);
+ self.metrics.row_replacements.add(1);
+ }
+ }
+ }
+ self.heap.insert_batch_entry(batch_entry);
+
+ // update memory reservation
+ self.reservation.try_resize(self.size())?;
+ Ok(())
+ }
+
+ /// Returns the top k results broken into `batch_size` [`RecordBatch`]es
+ pub fn emit(self) -> Result<SendableRecordBatchStream> {
+ let Self {
+ schema,
+ metrics,
+ reservation: _,
+ batch_size,
+ expr: _,
+ row_converter: _,
+ scratch_rows: _,
+ heap,
+ } = self;
+ let _timer = metrics.baseline.elapsed_compute().timer(); // time updated on drop
+
+ let mut batch = heap.emit(schema.clone())?;
+ metrics.baseline.output_rows().add(batch.num_rows());
+
+ // break into record batches as needed
+ let mut batches = vec![];
+ loop {
+ if batch.num_rows() < batch_size {
+ batches.push(Ok(batch));
+ break;
+ } else {
+ batches.push(Ok(batch.slice(0, batch_size)));
+ batch = batch.slice(batch_size, batch.num_rows());
+ }
+ }
+ Ok(Box::pin(RecordBatchStreamAdapter::new(
+ schema,
+ futures::stream::iter(batches),
+ )))
+ }
+
+ /// return the size of memory used by this operator, in bytes
+ fn size(&self) -> usize {
+ std::mem::size_of::<Self>()
+ + self.row_converter.size()
+ + self.scratch_rows.size()
+ + self.heap.size()
+ }
+}
+
+struct TopKMetrics {
+ /// metrics
+ pub baseline: BaselineMetrics,
+
+ /// count of how many rows were replaced in the heap
+ pub row_replacements: Count,
+}
+
+impl TopKMetrics {
+ fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
+ Self {
+ baseline: BaselineMetrics::new(metrics, partition),
+ row_replacements: MetricBuilder::new(metrics)
+ .counter("row_replacements", partition),
+ }
+ }
+}
+
+/// This structure keeps at most the *smallest* k items, using the
+/// [arrow::row] format for sort keys. While it is called "topK" for
+/// values like `1, 2, 3, 4, 5` the "top 3" really means the
+/// *smallest* 3 , `1, 2, 3`, not the *largest* 3 `3, 4, 5`.
+///
+/// Using the `Row` format handles things such as ascending vs
+/// descending and nulls first vs nulls last.
+///
+/// It doesn't use `BinaryHeap` in the Rust standard library because
+/// it is important to check the current minimum value in the heap
+/// prior to creating a new value to insert.
+struct TopKHeap {
+ /// The maximum size of this heap.
+ k: usize,
+ /// Storage for up at most `k` items, in ascending
+ /// order. `inner[0]` holds the smallest value of the smallest k
+ /// so far, `inner[len-1]` holds the largest value smallest k so far.
+ inner: Vec<TopKRow>,
Review Comment:
I feel like there must be some more clever algorithmic trick to keep a sorted list / balanced tree but this one seems to work pretty well 🤷 I do think it will have `O(N * K)` worst case performance (if the input is reverse ordered so every new values is a new topK).
##########
datafusion/core/src/physical_plan/topk/mod.rs:
##########
@@ -0,0 +1,516 @@
+// 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.
+
+//! TopK: Combination of Sort / LIMIT
+
+use arrow::{
+ compute::interleave,
+ row::{OwnedRow, RowConverter, Rows, SortField},
+};
+use std::{cmp::Ordering, sync::Arc};
+
+use arrow_array::{Array, ArrayRef, RecordBatch};
+use arrow_schema::SchemaRef;
+use datafusion_common::Result;
+use datafusion_execution::{
+ memory_pool::{MemoryConsumer, MemoryReservation},
+ runtime_env::RuntimeEnv,
+};
+use datafusion_physical_expr::PhysicalSortExpr;
+use hashbrown::HashMap;
+
+use crate::physical_plan::{stream::RecordBatchStreamAdapter, SendableRecordBatchStream};
+
+use super::metrics::{BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder};
+
+/// Global TopK
+///
+/// # Background
+///
+/// "Top K" is a common query optimization used for queries such as
+/// "find the top 3 customers by revenue". The (simplified) SQL for
+/// such a query might be:
+///
+/// ```sql
+/// SELECT customer_id, revenue FROM 'sales.csv' ORDER BY revenue DESC limit 3;
+/// ```
+///
+/// The simple plan would be:
+///
+/// ```
+/// > explain SELECT customer_id, revenue FROM sales ORDER BY revenue DESC limit 3;
+/// +--------------+----------------------------------------+
+/// | plan_type | plan |
+/// +--------------+----------------------------------------+
+/// | logical_plan | Limit: 3 |
+/// | | Sort: revenue DESC NULLS FIRST |
+/// | | Projection: customer_id, revenue |
+/// | | TableScan: sales |
+/// +--------------+----------------------------------------+
+/// ```
+///
+/// While this plan produces the correct answer, it will fully sorts the
+/// input before discarding everything other than the top 3 elements.
+///
+/// The same answer can be produced by simply keeping track of the top
+/// N elements, reducing the total amount of required buffer memory.
+///
+/// # Structure
+///
+/// This operator tracks the top K items using a `TopKHeap`.
+pub struct TopK {
+ /// schema of the output (and the input)
+ schema: SchemaRef,
+ /// Runtime metrics
+ metrics: TopKMetrics,
+ /// Reservation
+ reservation: MemoryReservation,
+ /// The target number of rows for output batches
+ batch_size: usize,
+ /// sort expressions
+ expr: Arc<[PhysicalSortExpr]>,
+ /// row converter, for sort keys
+ row_converter: RowConverter,
+ /// scratch space for converting rows
+ scratch_rows: Rows,
+ /// stores the top k values and their sort key values, in order
+ heap: TopKHeap,
+}
+
+impl TopK {
+ /// Create a new [`TopK`] that stores the top `k` values, as
+ /// defined by the sort expressions in `expr`.
+ // TOOD: make a builder or some other nicer API to avoid the
+ // clippy warning
+ #[allow(clippy::too_many_arguments)]
+ pub fn try_new(
+ partition_id: usize,
+ schema: SchemaRef,
+ expr: Vec<PhysicalSortExpr>,
+ k: usize,
+ batch_size: usize,
+ runtime: Arc<RuntimeEnv>,
+ metrics: &ExecutionPlanMetricsSet,
+ partition: usize,
+ ) -> Result<Self> {
+ let reservation = MemoryConsumer::new(format!("TopK[{partition_id}]"))
+ .register(&runtime.memory_pool);
+
+ let expr: Arc<[PhysicalSortExpr]> = expr.into();
+
+ let sort_fields: Vec<_> = expr
+ .iter()
+ .map(|e| {
+ Ok(SortField::new_with_options(
+ e.expr.data_type(&schema)?,
+ e.options,
+ ))
+ })
+ .collect::<Result<_>>()?;
+
+ let row_converter = RowConverter::new(sort_fields)?;
+ let scratch_rows = row_converter.empty_rows(
+ batch_size,
+ 20 * batch_size, // guestimate 20 bytes per row
+ );
+
+ Ok(Self {
+ schema,
+ metrics: TopKMetrics::new(metrics, partition),
+ reservation,
+ batch_size,
+ expr,
+ row_converter,
+ scratch_rows,
+ heap: TopKHeap::new(k),
+ })
+ }
+
+ /// Insert `batch`, remembering it if any of its values are among
+ /// the top k seen so far.
+ pub fn insert_batch(&mut self, batch: RecordBatch) -> Result<()> {
+ // Updates on drop
+ let _timer = self.metrics.baseline.elapsed_compute().timer();
+
+ let sort_keys: Vec<ArrayRef> = self
+ .expr
+ .iter()
+ .map(|expr| {
+ let value = expr.expr.evaluate(&batch)?;
+ Ok(value.into_array(batch.num_rows()))
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // reuse existing `Rows` to avoid reallocations
+ let rows = &mut self.scratch_rows;
+ rows.clear();
+ self.row_converter.append(rows, &sort_keys)?;
+
+ // TODO make this algorithmically better?:
+ // 1. only check topk values in rows
+ // 2. only do one update through top_k
+
+ let mut batch_entry = self.heap.register_batch(batch);
+ for (index, row) in rows.iter().enumerate() {
Review Comment:
The is the core of the algorithm -- it is pretty unfancy but avoids allocations unless there is an actual new top value. I will file some potential arrow-rs improvements to make this even faster
##########
datafusion/core/tests/sqllogictests/test_files/window.slt:
##########
@@ -2642,8 +2642,9 @@ ProjectionExec: expr=[fv1@0 as fv1, fv2@1 as fv2, lv1@2 as lv1, lv2@3 as lv2, nv
----------BoundedWindowAggExec: wdw=[FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(1)), end_bound: Following(Int32(10)) }, FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound: Follow
ing(UInt64(10)) }, LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(1)), end_bound: Following(Int32(10)) }, LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound: Following(UInt64(10)) }, LAG
(annotated_data_finite.inc_col,Int64(1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING: Ok(Field { name: "LAG(annotated_data_finite.inc_col,Int64(1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(10)), end_bound: Following(Int32(1)) }, LAG(annotated_data_finite.inc_col,Int64(2),Int64(1002)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LAG(annotated_data_finite.inc_col,Int64(2),Int64(1002)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_b
ound: Following(UInt64(10)) }, LEAD(annotated_data_finite.inc_col,Int64(-1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING: Ok(Field { name: "LEAD(annotated_data_finite.inc_col,Int64(-1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(10)), end_bound: Following(Int32(1)) }, LEAD(annotated_data_finite.inc_col,Int64(4),Int64(1004)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LEAD(annotated_data_finite.inc_col,Int64(4),Int64(1004)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows,
start_bound: Preceding(UInt64(1)), end_bound: Following(UInt64(10)) }], mode=[Sorted]
------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_1.csv]]}, projection=[ts, inc_col], output_ordering=[ts@0 ASC NULLS LAST], has_header=true
-query IIIIIIIIIIIIIIIIIIIIIIII
+query IIIIIIIIIIIIIIIIIIIIIIIII
SELECT
+ ts,
Review Comment:
I added `ts` to this query to show that the the first two rows have a tie in the same value `264` and thus it is ok that the output order changes. We should probably fix the test (or could add `rowsort`) but that might obscure real errors in the future 🤔
##########
datafusion/core/src/physical_plan/topk/mod.rs:
##########
@@ -0,0 +1,516 @@
+// 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.
+
+//! TopK: Combination of Sort / LIMIT
+
+use arrow::{
+ compute::interleave,
+ row::{OwnedRow, RowConverter, Rows, SortField},
+};
+use std::{cmp::Ordering, sync::Arc};
+
+use arrow_array::{Array, ArrayRef, RecordBatch};
+use arrow_schema::SchemaRef;
+use datafusion_common::Result;
+use datafusion_execution::{
+ memory_pool::{MemoryConsumer, MemoryReservation},
+ runtime_env::RuntimeEnv,
+};
+use datafusion_physical_expr::PhysicalSortExpr;
+use hashbrown::HashMap;
+
+use crate::physical_plan::{stream::RecordBatchStreamAdapter, SendableRecordBatchStream};
+
+use super::metrics::{BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder};
+
+/// Global TopK
+///
+/// # Background
+///
+/// "Top K" is a common query optimization used for queries such as
+/// "find the top 3 customers by revenue". The (simplified) SQL for
+/// such a query might be:
+///
+/// ```sql
+/// SELECT customer_id, revenue FROM 'sales.csv' ORDER BY revenue DESC limit 3;
+/// ```
+///
+/// The simple plan would be:
+///
+/// ```
+/// > explain SELECT customer_id, revenue FROM sales ORDER BY revenue DESC limit 3;
+/// +--------------+----------------------------------------+
+/// | plan_type | plan |
+/// +--------------+----------------------------------------+
+/// | logical_plan | Limit: 3 |
+/// | | Sort: revenue DESC NULLS FIRST |
+/// | | Projection: customer_id, revenue |
+/// | | TableScan: sales |
+/// +--------------+----------------------------------------+
+/// ```
+///
+/// While this plan produces the correct answer, it will fully sorts the
+/// input before discarding everything other than the top 3 elements.
+///
+/// The same answer can be produced by simply keeping track of the top
+/// N elements, reducing the total amount of required buffer memory.
+///
+/// # Structure
+///
+/// This operator tracks the top K items using a `TopKHeap`.
+pub struct TopK {
+ /// schema of the output (and the input)
+ schema: SchemaRef,
+ /// Runtime metrics
+ metrics: TopKMetrics,
+ /// Reservation
+ reservation: MemoryReservation,
+ /// The target number of rows for output batches
+ batch_size: usize,
+ /// sort expressions
+ expr: Arc<[PhysicalSortExpr]>,
+ /// row converter, for sort keys
+ row_converter: RowConverter,
+ /// scratch space for converting rows
+ scratch_rows: Rows,
+ /// stores the top k values and their sort key values, in order
+ heap: TopKHeap,
+}
+
+impl TopK {
+ /// Create a new [`TopK`] that stores the top `k` values, as
+ /// defined by the sort expressions in `expr`.
+ // TOOD: make a builder or some other nicer API to avoid the
+ // clippy warning
+ #[allow(clippy::too_many_arguments)]
+ pub fn try_new(
+ partition_id: usize,
+ schema: SchemaRef,
+ expr: Vec<PhysicalSortExpr>,
+ k: usize,
+ batch_size: usize,
+ runtime: Arc<RuntimeEnv>,
+ metrics: &ExecutionPlanMetricsSet,
+ partition: usize,
+ ) -> Result<Self> {
+ let reservation = MemoryConsumer::new(format!("TopK[{partition_id}]"))
+ .register(&runtime.memory_pool);
+
+ let expr: Arc<[PhysicalSortExpr]> = expr.into();
+
+ let sort_fields: Vec<_> = expr
+ .iter()
+ .map(|e| {
+ Ok(SortField::new_with_options(
+ e.expr.data_type(&schema)?,
+ e.options,
+ ))
+ })
+ .collect::<Result<_>>()?;
+
+ let row_converter = RowConverter::new(sort_fields)?;
+ let scratch_rows = row_converter.empty_rows(
+ batch_size,
+ 20 * batch_size, // guestimate 20 bytes per row
+ );
+
+ Ok(Self {
+ schema,
+ metrics: TopKMetrics::new(metrics, partition),
+ reservation,
+ batch_size,
+ expr,
+ row_converter,
+ scratch_rows,
+ heap: TopKHeap::new(k),
+ })
+ }
+
+ /// Insert `batch`, remembering it if any of its values are among
+ /// the top k seen so far.
+ pub fn insert_batch(&mut self, batch: RecordBatch) -> Result<()> {
+ // Updates on drop
+ let _timer = self.metrics.baseline.elapsed_compute().timer();
+
+ let sort_keys: Vec<ArrayRef> = self
+ .expr
+ .iter()
+ .map(|expr| {
+ let value = expr.expr.evaluate(&batch)?;
+ Ok(value.into_array(batch.num_rows()))
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // reuse existing `Rows` to avoid reallocations
+ let rows = &mut self.scratch_rows;
+ rows.clear();
+ self.row_converter.append(rows, &sort_keys)?;
+
+ // TODO make this algorithmically better?:
+ // 1. only check topk values in rows
+ // 2. only do one update through top_k
+
+ let mut batch_entry = self.heap.register_batch(batch);
+ for (index, row) in rows.iter().enumerate() {
+ match self.heap.k_largest() {
+ // heap has k items, and the current row is not
+ // smaller than the curret smallest k value, skip
+ Some(largest) if largest.row.row() <= row => {}
+ // don't yet have k items or new item is greater than
+ // current min top k
+ None | Some(_) => {
+ self.heap.add(&mut batch_entry, row.owned(), index);
+ self.metrics.row_replacements.add(1);
+ }
+ }
+ }
+ self.heap.insert_batch_entry(batch_entry);
+
+ // update memory reservation
+ self.reservation.try_resize(self.size())?;
+ Ok(())
+ }
+
+ /// Returns the top k results broken into `batch_size` [`RecordBatch`]es
+ pub fn emit(self) -> Result<SendableRecordBatchStream> {
+ let Self {
+ schema,
+ metrics,
+ reservation: _,
+ batch_size,
+ expr: _,
+ row_converter: _,
+ scratch_rows: _,
+ heap,
+ } = self;
+ let _timer = metrics.baseline.elapsed_compute().timer(); // time updated on drop
+
+ let mut batch = heap.emit(schema.clone())?;
+ metrics.baseline.output_rows().add(batch.num_rows());
+
+ // break into record batches as needed
+ let mut batches = vec![];
+ loop {
+ if batch.num_rows() < batch_size {
+ batches.push(Ok(batch));
+ break;
+ } else {
+ batches.push(Ok(batch.slice(0, batch_size)));
+ batch = batch.slice(batch_size, batch.num_rows());
+ }
+ }
+ Ok(Box::pin(RecordBatchStreamAdapter::new(
+ schema,
+ futures::stream::iter(batches),
+ )))
+ }
+
+ /// return the size of memory used by this operator, in bytes
+ fn size(&self) -> usize {
+ std::mem::size_of::<Self>()
+ + self.row_converter.size()
+ + self.scratch_rows.size()
+ + self.heap.size()
+ }
+}
+
+struct TopKMetrics {
+ /// metrics
+ pub baseline: BaselineMetrics,
+
+ /// count of how many rows were replaced in the heap
+ pub row_replacements: Count,
+}
+
+impl TopKMetrics {
+ fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
+ Self {
+ baseline: BaselineMetrics::new(metrics, partition),
+ row_replacements: MetricBuilder::new(metrics)
+ .counter("row_replacements", partition),
+ }
+ }
+}
+
+/// This structure keeps at most the *smallest* k items, using the
+/// [arrow::row] format for sort keys. While it is called "topK" for
+/// values like `1, 2, 3, 4, 5` the "top 3" really means the
+/// *smallest* 3 , `1, 2, 3`, not the *largest* 3 `3, 4, 5`.
+///
+/// Using the `Row` format handles things such as ascending vs
+/// descending and nulls first vs nulls last.
+///
+/// It doesn't use `BinaryHeap` in the Rust standard library because
+/// it is important to check the current minimum value in the heap
+/// prior to creating a new value to insert.
+struct TopKHeap {
+ /// The maximum size of this heap.
+ k: usize,
+ /// Storage for up at most `k` items, in ascending
+ /// order. `inner[0]` holds the smallest value of the smallest k
+ /// so far, `inner[len-1]` holds the largest value smallest k so far.
+ inner: Vec<TopKRow>,
Review Comment:
I feel like there must be some more clever algorithmic trick to keep a sorted list / balanced tree but this one seems to work pretty well 🤷 I do think it will have `O(N * K)` worst case performance (if the input is reverse ordered so every new values is a new topK).
##########
datafusion/core/tests/sqllogictests/test_files/window.slt:
##########
@@ -2642,8 +2642,9 @@ ProjectionExec: expr=[fv1@0 as fv1, fv2@1 as fv2, lv1@2 as lv1, lv2@3 as lv2, nv
----------BoundedWindowAggExec: wdw=[FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(1)), end_bound: Following(Int32(10)) }, FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "FIRST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound: Follow
ing(UInt64(10)) }, LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(1)), end_bound: Following(Int32(10)) }, LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LAST_VALUE(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound: Following(UInt64(10)) }, LAG
(annotated_data_finite.inc_col,Int64(1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING: Ok(Field { name: "LAG(annotated_data_finite.inc_col,Int64(1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(10)), end_bound: Following(Int32(1)) }, LAG(annotated_data_finite.inc_col,Int64(2),Int64(1002)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LAG(annotated_data_finite.inc_col,Int64(2),Int64(1002)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_b
ound: Following(UInt64(10)) }, LEAD(annotated_data_finite.inc_col,Int64(-1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING: Ok(Field { name: "LEAD(annotated_data_finite.inc_col,Int64(-1),Int64(1001)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 1 PRECEDING AND 10 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(10)), end_bound: Following(Int32(1)) }, LEAD(annotated_data_finite.inc_col,Int64(4),Int64(1004)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING: Ok(Field { name: "LEAD(annotated_data_finite.inc_col,Int64(4),Int64(1004)) ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows,
start_bound: Preceding(UInt64(1)), end_bound: Following(UInt64(10)) }], mode=[Sorted]
------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_1.csv]]}, projection=[ts, inc_col], output_ordering=[ts@0 ASC NULLS LAST], has_header=true
-query IIIIIIIIIIIIIIIIIIIIIIII
+query IIIIIIIIIIIIIIIIIIIIIIIII
SELECT
+ ts,
Review Comment:
I added `ts` to this query to show that the the first two rows have a tie in the same value `264` and thus it is ok that the output order changes. We should probably fix the test (or could add `rowsort`) but that might obscure real errors in the future 🤔
##########
datafusion/core/tests/sqllogictests/test_files/aal.slt:
##########
@@ -0,0 +1,202 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
this is for testing -- it would not be checked in to a final PR
##########
datafusion/core/tests/sqllogictests/test_files/window.slt:
##########
@@ -2597,6 +2597,7 @@ SELECT
# test_source_sorted_builtin
query TT
EXPLAIN SELECT
+ ts,
Review Comment:
this change is to demonstrate that the `ts` column is not unique in the first `5` values and thus that a different output order is acceptable
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org