You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ag...@apache.org on 2022/09/07 14:50:53 UTC
[arrow-datafusion] branch master updated: Use `usize` rather than `Option` to represent `Limit::skip`and `Limit::offset` (#3374)
This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git
The following commit(s) were added to refs/heads/master by this push:
new 43e2d91c7 Use `usize` rather than `Option<usize>` to represent `Limit::skip`and `Limit::offset` (#3374)
43e2d91c7 is described below
commit 43e2d91c78460d0670037053a23e6d3b2f35d42d
Author: Remzi Yang <59...@users.noreply.github.com>
AuthorDate: Wed Sep 7 22:50:47 2022 +0800
Use `usize` rather than `Option<usize>` to represent `Limit::skip`and `Limit::offset` (#3374)
---
datafusion/core/src/dataframe.rs | 32 +++---
datafusion/core/src/execution/context.rs | 2 +-
datafusion/core/src/lib.rs | 2 +-
.../core/src/physical_optimizer/repartition.rs | 16 +--
datafusion/core/src/physical_plan/limit.rs | 50 ++++-----
datafusion/core/src/physical_plan/planner.rs | 16 +--
datafusion/core/tests/dataframe_functions.rs | 2 +-
datafusion/core/tests/sql/explain_analyze.rs | 4 +-
datafusion/expr/src/logical_plan/builder.rs | 4 +-
datafusion/expr/src/logical_plan/plan.rs | 7 +-
datafusion/optimizer/src/eliminate_limit.rs | 41 ++++----
datafusion/optimizer/src/filter_push_down.rs | 24 ++---
datafusion/optimizer/src/limit_push_down.rs | 113 ++++++++++-----------
datafusion/optimizer/src/projection_push_down.rs | 4 +-
datafusion/optimizer/src/simplify_expressions.rs | 4 +-
datafusion/proto/src/logical_plan.rs | 8 +-
datafusion/sql/src/planner.rs | 39 ++++---
docs/source/user-guide/dataframe.md | 2 +-
18 files changed, 173 insertions(+), 197 deletions(-)
diff --git a/datafusion/core/src/dataframe.rs b/datafusion/core/src/dataframe.rs
index 3c77f1671..ea90a9b81 100644
--- a/datafusion/core/src/dataframe.rs
+++ b/datafusion/core/src/dataframe.rs
@@ -62,7 +62,7 @@ use std::sync::Arc;
/// let df = ctx.read_csv("tests/example.csv", CsvReadOptions::new()).await?;
/// let df = df.filter(col("a").lt_eq(col("b")))?
/// .aggregate(vec![col("a")], vec![min(col("b"))])?
-/// .limit(None, Some(100))?;
+/// .limit(0, Some(100))?;
/// let results = df.collect();
/// # Ok(())
/// # }
@@ -217,15 +217,11 @@ impl DataFrame {
/// # async fn main() -> Result<()> {
/// let ctx = SessionContext::new();
/// let df = ctx.read_csv("tests/example.csv", CsvReadOptions::new()).await?;
- /// let df = df.limit(None, Some(100))?;
+ /// let df = df.limit(0, Some(100))?;
/// # Ok(())
/// # }
/// ```
- pub fn limit(
- &self,
- skip: Option<usize>,
- fetch: Option<usize>,
- ) -> Result<Arc<DataFrame>> {
+ pub fn limit(&self, skip: usize, fetch: Option<usize>) -> Result<Arc<DataFrame>> {
let plan = LogicalPlanBuilder::from(self.plan.clone())
.limit(skip, fetch)?
.build()?;
@@ -438,7 +434,7 @@ impl DataFrame {
/// # }
/// ```
pub async fn show_limit(&self, num: usize) -> Result<()> {
- let results = self.limit(None, Some(num))?.collect().await?;
+ let results = self.limit(0, Some(num))?.collect().await?;
Ok(pretty::print_batches(&results)?)
}
@@ -543,7 +539,7 @@ impl DataFrame {
/// # async fn main() -> Result<()> {
/// let ctx = SessionContext::new();
/// let df = ctx.read_csv("tests/example.csv", CsvReadOptions::new()).await?;
- /// let batches = df.limit(None, Some(100))?.explain(false, false)?.collect().await?;
+ /// let batches = df.limit(0, Some(100))?.explain(false, false)?.collect().await?;
/// # Ok(())
/// # }
/// ```
@@ -789,7 +785,7 @@ impl TableProvider for DataFrame {
Self::new(
self.session_state.clone(),
&limit
- .map_or_else(|| Ok(expr.clone()), |n| expr.limit(None, Some(n)))?
+ .map_or_else(|| Ok(expr.clone()), |n| expr.limit(0, Some(n)))?
.plan
.clone(),
)
@@ -923,9 +919,7 @@ mod tests {
async fn limit() -> Result<()> {
// build query using Table API
let t = test_table().await?;
- let t2 = t
- .select_columns(&["c1", "c2", "c11"])?
- .limit(None, Some(10))?;
+ let t2 = t.select_columns(&["c1", "c2", "c11"])?.limit(0, Some(10))?;
let plan = t2.plan.clone();
// build query using SQL
@@ -944,7 +938,7 @@ mod tests {
let df = test_table().await?;
let df = df
.select_columns(&["c1", "c2", "c11"])?
- .limit(None, Some(10))?
+ .limit(0, Some(10))?
.explain(false, false)?;
let plan = df.plan.clone();
@@ -1205,7 +1199,7 @@ mod tests {
.await?
.select_columns(&["c1", "c2", "c3"])?
.filter(col("c2").eq(lit(3)).and(col("c1").eq(lit("a"))))?
- .limit(None, Some(1))?
+ .limit(0, Some(1))?
.sort(vec![
// make the test deterministic
col("c1").sort(true, true),
@@ -1248,7 +1242,7 @@ mod tests {
col("t2.c2").sort(true, true),
col("t2.c3").sort(true, true),
])?
- .limit(None, Some(1))?;
+ .limit(0, Some(1))?;
let df_results = df.collect().await?;
assert_batches_sorted_eq!(
@@ -1266,7 +1260,7 @@ mod tests {
assert_eq!("\
Projection: #t1.c1 AS AAA, #t1.c2, #t1.c3, #t2.c1, #t2.c2, #t2.c3\
- \n Limit: skip=None, fetch=1\
+ \n Limit: skip=0, fetch=1\
\n Sort: #t1.c1 ASC NULLS FIRST, #t1.c2 ASC NULLS FIRST, #t1.c3 ASC NULLS FIRST, #t2.c1 ASC NULLS FIRST, #t2.c2 ASC NULLS FIRST, #t2.c3 ASC NULLS FIRST\
\n Inner Join: #t1.c1 = #t2.c1\
\n TableScan: t1\
@@ -1276,7 +1270,7 @@ mod tests {
assert_eq!("\
Projection: #t1.c1 AS AAA, #t1.c2, #t1.c3, #t2.c1, #t2.c2, #t2.c3\
- \n Limit: skip=None, fetch=1\
+ \n Limit: skip=0, fetch=1\
\n Sort: #t1.c1 ASC NULLS FIRST, #t1.c2 ASC NULLS FIRST, #t1.c3 ASC NULLS FIRST, #t2.c1 ASC NULLS FIRST, #t2.c2 ASC NULLS FIRST, #t2.c3 ASC NULLS FIRST\
\n Inner Join: #t1.c1 = #t2.c1\
\n TableScan: t1 projection=[c1, c2, c3]\
@@ -1305,7 +1299,7 @@ mod tests {
let df = test_table()
.await?
.select_columns(&["c2", "c3"])?
- .limit(None, Some(1))?
+ .limit(0, Some(1))?
.with_column("sum", cast(col("c2") + col("c3"), DataType::Int64))?;
let df_results = df.collect().await?;
diff --git a/datafusion/core/src/execution/context.rs b/datafusion/core/src/execution/context.rs
index a6a508d36..dd67f5a63 100644
--- a/datafusion/core/src/execution/context.rs
+++ b/datafusion/core/src/execution/context.rs
@@ -149,7 +149,7 @@ const DEFAULT_SCHEMA: &str = "public";
/// let df = ctx.read_csv("tests/example.csv", CsvReadOptions::new()).await?;
/// let df = df.filter(col("a").lt_eq(col("b")))?
/// .aggregate(vec![col("a")], vec![min(col("b"))])?
-/// .limit(None, Some(100))?;
+/// .limit(0, Some(100))?;
/// let results = df.collect();
/// # Ok(())
/// # }
diff --git a/datafusion/core/src/lib.rs b/datafusion/core/src/lib.rs
index f5b8cf431..790f03411 100644
--- a/datafusion/core/src/lib.rs
+++ b/datafusion/core/src/lib.rs
@@ -44,7 +44,7 @@
//! // create a plan
//! let df = df.filter(col("a").lt_eq(col("b")))?
//! .aggregate(vec![col("a")], vec![min(col("b"))])?
-//! .limit(None, Some(100))?;
+//! .limit(0, Some(100))?;
//!
//! // execute the plan
//! let results: Vec<RecordBatch> = df.collect().await?;
diff --git a/datafusion/core/src/physical_optimizer/repartition.rs b/datafusion/core/src/physical_optimizer/repartition.rs
index ec2a1355f..52e36d8d5 100644
--- a/datafusion/core/src/physical_optimizer/repartition.rs
+++ b/datafusion/core/src/physical_optimizer/repartition.rs
@@ -329,7 +329,7 @@ mod tests {
fn limit_exec(input: Arc<dyn ExecutionPlan>) -> Arc<dyn ExecutionPlan> {
Arc::new(GlobalLimitExec::new(
Arc::new(LocalLimitExec::new(input, 100)),
- None,
+ 0,
Some(100),
))
}
@@ -337,7 +337,7 @@ mod tests {
fn limit_exec_with_skip(input: Arc<dyn ExecutionPlan>) -> Arc<dyn ExecutionPlan> {
Arc::new(GlobalLimitExec::new(
Arc::new(LocalLimitExec::new(input, 100)),
- Some(5),
+ 5,
Some(100),
))
}
@@ -407,7 +407,7 @@ mod tests {
let plan = limit_exec(filter_exec(parquet_exec()));
let expected = &[
- "GlobalLimitExec: skip=None, fetch=100",
+ "GlobalLimitExec: skip=0, fetch=100",
"LocalLimitExec: fetch=100",
"FilterExec: c1@0",
// nothing sorts the data, so the local limit doesn't require sorted data either
@@ -441,7 +441,7 @@ mod tests {
let plan = limit_exec(sort_exec(parquet_exec()));
let expected = &[
- "GlobalLimitExec: skip=None, fetch=100",
+ "GlobalLimitExec: skip=0, fetch=100",
"LocalLimitExec: fetch=100",
// data is sorted so can't repartition here
"SortExec: [c1@0 ASC]",
@@ -457,7 +457,7 @@ mod tests {
let plan = limit_exec(filter_exec(sort_exec(parquet_exec())));
let expected = &[
- "GlobalLimitExec: skip=None, fetch=100",
+ "GlobalLimitExec: skip=0, fetch=100",
"LocalLimitExec: fetch=100",
"FilterExec: c1@0",
// data is sorted so can't repartition here even though
@@ -478,12 +478,12 @@ mod tests {
"AggregateExec: mode=Final, gby=[], aggr=[]",
"AggregateExec: mode=Partial, gby=[], aggr=[]",
"RepartitionExec: partitioning=RoundRobinBatch(10)",
- "GlobalLimitExec: skip=None, fetch=100",
+ "GlobalLimitExec: skip=0, fetch=100",
"LocalLimitExec: fetch=100",
"FilterExec: c1@0",
// repartition should happen prior to the filter to maximize parallelism
"RepartitionExec: partitioning=RoundRobinBatch(10)",
- "GlobalLimitExec: skip=None, fetch=100",
+ "GlobalLimitExec: skip=0, fetch=100",
"LocalLimitExec: fetch=100",
// Expect no repartition to happen for local limit
"ParquetExec: limit=None, partitions=[x], projection=[c1]",
@@ -508,7 +508,7 @@ mod tests {
"FilterExec: c1@0",
// repartition should happen prior to the filter to maximize parallelism
"RepartitionExec: partitioning=RoundRobinBatch(10)",
- "GlobalLimitExec: skip=None, fetch=100",
+ "GlobalLimitExec: skip=0, fetch=100",
"LocalLimitExec: fetch=100",
// Expect no repartition to happen for local limit
"ParquetExec: limit=None, partitions=[x], projection=[c1]",
diff --git a/datafusion/core/src/physical_plan/limit.rs b/datafusion/core/src/physical_plan/limit.rs
index e9ff61409..322c21ff4 100644
--- a/datafusion/core/src/physical_plan/limit.rs
+++ b/datafusion/core/src/physical_plan/limit.rs
@@ -49,8 +49,9 @@ pub struct GlobalLimitExec {
/// Input execution plan
input: Arc<dyn ExecutionPlan>,
/// Number of rows to skip before fetch
- skip: Option<usize>,
- /// Maximum number of rows to fetch
+ skip: usize,
+ /// Maximum number of rows to fetch,
+ /// `None` means fetching all rows
fetch: Option<usize>,
/// Execution metrics
metrics: ExecutionPlanMetricsSet,
@@ -58,11 +59,7 @@ pub struct GlobalLimitExec {
impl GlobalLimitExec {
/// Create a new GlobalLimitExec
- pub fn new(
- input: Arc<dyn ExecutionPlan>,
- skip: Option<usize>,
- fetch: Option<usize>,
- ) -> Self {
+ pub fn new(input: Arc<dyn ExecutionPlan>, skip: usize, fetch: Option<usize>) -> Self {
GlobalLimitExec {
input,
skip,
@@ -77,8 +74,8 @@ impl GlobalLimitExec {
}
/// Number of rows to skip before fetch
- pub fn skip(&self) -> Option<&usize> {
- self.skip.as_ref()
+ pub fn skip(&self) -> usize {
+ self.skip
}
/// Maximum number of rows to fetch
@@ -181,7 +178,7 @@ impl ExecutionPlan for GlobalLimitExec {
write!(
f,
"GlobalLimitExec: skip={}, fetch={}",
- self.skip.map_or("None".to_string(), |x| x.to_string()),
+ self.skip,
self.fetch.map_or("None".to_string(), |x| x.to_string())
)
}
@@ -194,7 +191,7 @@ impl ExecutionPlan for GlobalLimitExec {
fn statistics(&self) -> Statistics {
let input_stats = self.input.statistics();
- let skip = self.skip.unwrap_or(0);
+ let skip = self.skip;
match input_stats {
Statistics {
num_rows: Some(nr), ..
@@ -319,7 +316,7 @@ impl ExecutionPlan for LocalLimitExec {
let stream = self.input.execute(partition, context)?;
Ok(Box::pin(LimitStream::new(
stream,
- None,
+ 0,
Some(self.fetch),
baseline_metrics,
)))
@@ -397,13 +394,13 @@ struct LimitStream {
impl LimitStream {
fn new(
input: SendableRecordBatchStream,
- skip: Option<usize>,
+ skip: usize,
fetch: Option<usize>,
baseline_metrics: BaselineMetrics,
) -> Self {
let schema = input.schema();
Self {
- skip: skip.unwrap_or(0),
+ skip,
fetch: fetch.unwrap_or(usize::MAX),
input: Some(input),
schema,
@@ -524,11 +521,8 @@ mod tests {
// input should have 4 partitions
assert_eq!(csv.output_partitioning().partition_count(), num_partitions);
- let limit = GlobalLimitExec::new(
- Arc::new(CoalescePartitionsExec::new(csv)),
- None,
- Some(7),
- );
+ let limit =
+ GlobalLimitExec::new(Arc::new(CoalescePartitionsExec::new(csv)), 0, Some(7));
// the result should contain 4 batches (one per input partition)
let iter = limit.execute(0, task_ctx)?;
@@ -559,7 +553,7 @@ mod tests {
// (5 rows) and 1 row from the second (1 row)
let baseline_metrics = BaselineMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
let limit_stream =
- LimitStream::new(Box::pin(input), None, Some(6), baseline_metrics);
+ LimitStream::new(Box::pin(input), 0, Some(6), baseline_metrics);
assert_eq!(index.value(), 0);
let results = collect(Box::pin(limit_stream)).await.unwrap();
@@ -574,7 +568,7 @@ mod tests {
}
// test cases for "skip"
- async fn skip_and_fetch(skip: Option<usize>, fetch: Option<usize>) -> Result<usize> {
+ async fn skip_and_fetch(skip: usize, fetch: Option<usize>) -> Result<usize> {
let session_ctx = SessionContext::new();
let task_ctx = session_ctx.task_ctx();
@@ -594,14 +588,14 @@ mod tests {
#[tokio::test]
async fn skip_none_fetch_none() -> Result<()> {
- let row_count = skip_and_fetch(None, None).await?;
+ let row_count = skip_and_fetch(0, None).await?;
assert_eq!(row_count, 100);
Ok(())
}
#[tokio::test]
async fn skip_none_fetch_50() -> Result<()> {
- let row_count = skip_and_fetch(None, Some(50)).await?;
+ let row_count = skip_and_fetch(0, Some(50)).await?;
assert_eq!(row_count, 50);
Ok(())
}
@@ -609,7 +603,7 @@ mod tests {
#[tokio::test]
async fn skip_3_fetch_none() -> Result<()> {
// there are total of 100 rows, we skipped 3 rows (offset = 3)
- let row_count = skip_and_fetch(Some(3), None).await?;
+ let row_count = skip_and_fetch(3, None).await?;
assert_eq!(row_count, 97);
Ok(())
}
@@ -617,21 +611,21 @@ mod tests {
#[tokio::test]
async fn skip_3_fetch_10() -> Result<()> {
// there are total of 100 rows, we skipped 3 rows (offset = 3)
- let row_count = skip_and_fetch(Some(3), Some(10)).await?;
+ let row_count = skip_and_fetch(3, Some(10)).await?;
assert_eq!(row_count, 10);
Ok(())
}
#[tokio::test]
async fn skip_100_fetch_none() -> Result<()> {
- let row_count = skip_and_fetch(Some(100), None).await?;
+ let row_count = skip_and_fetch(100, None).await?;
assert_eq!(row_count, 0);
Ok(())
}
#[tokio::test]
async fn skip_100_fetch_1() -> Result<()> {
- let row_count = skip_and_fetch(Some(100), Some(1)).await?;
+ let row_count = skip_and_fetch(100, Some(1)).await?;
assert_eq!(row_count, 0);
Ok(())
}
@@ -639,7 +633,7 @@ mod tests {
#[tokio::test]
async fn skip_101_fetch_none() -> Result<()> {
// there are total of 100 rows, we skipped 101 rows (offset = 3)
- let row_count = skip_and_fetch(Some(101), None).await?;
+ let row_count = skip_and_fetch(101, None).await?;
assert_eq!(row_count, 0);
Ok(())
}
diff --git a/datafusion/core/src/physical_plan/planner.rs b/datafusion/core/src/physical_plan/planner.rs
index a2d868361..6ec35a596 100644
--- a/datafusion/core/src/physical_plan/planner.rs
+++ b/datafusion/core/src/physical_plan/planner.rs
@@ -1007,7 +1007,7 @@ impl DefaultPhysicalPlanner {
// Apply a LocalLimitExec to each partition. The optimizer will also insert
// a CoalescePartitionsExec between the GlobalLimitExec and LocalLimitExec
if let Some(fetch) = fetch {
- Arc::new(LocalLimitExec::new(input, *fetch + skip.unwrap_or(0)))
+ Arc::new(LocalLimitExec::new(input, *fetch + skip))
} else {
input
}
@@ -1710,7 +1710,7 @@ mod tests {
.project(vec![col("c1"), col("c2")])?
.aggregate(vec![col("c1")], vec![sum(col("c2"))])?
.sort(vec![col("c1").sort(true, true)])?
- .limit(Some(3), Some(10))?
+ .limit(3, Some(10))?
.build()?;
let plan = plan(&logical_plan).await?;
@@ -1802,7 +1802,7 @@ mod tests {
let logical_plan = test_csv_scan()
.await?
.filter(col("c7").lt(col("c12")))?
- .limit(Some(3), None)?
+ .limit(3, None)?
.build()?;
let plan = plan(&logical_plan).await?;
@@ -1812,16 +1812,16 @@ mod tests {
let _expected = "predicate: BinaryExpr { left: TryCastExpr { expr: Column { name: \"c7\", index: 6 }, cast_type: Float64 }, op: Lt, right: Column { name: \"c12\", index: 11 } }";
let plan_debug_str = format!("{:?}", plan);
assert!(plan_debug_str.contains("GlobalLimitExec"));
- assert!(plan_debug_str.contains("skip: Some(3)"));
+ assert!(plan_debug_str.contains("skip: 3"));
Ok(())
}
#[tokio::test]
async fn test_with_zero_offset_plan() -> Result<()> {
- let logical_plan = test_csv_scan().await?.limit(Some(0), None)?.build()?;
+ let logical_plan = test_csv_scan().await?.limit(0, None)?.build()?;
let plan = plan(&logical_plan).await?;
assert!(format!("{:?}", plan).contains("GlobalLimitExec"));
- assert!(format!("{:?}", plan).contains("skip: Some(0)"));
+ assert!(format!("{:?}", plan).contains("skip: 0"));
Ok(())
}
@@ -1830,12 +1830,12 @@ mod tests {
let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
let logical_plan = scan_empty_with_partitions(Some("test"), &schema, None, 2)?
- .limit(Some(3), Some(5))?
+ .limit(3, Some(5))?
.build()?;
let plan = plan(&logical_plan).await?;
assert!(format!("{:?}", plan).contains("GlobalLimitExec"));
- assert!(format!("{:?}", plan).contains("skip: Some(3), fetch: Some(5)"));
+ assert!(format!("{:?}", plan).contains("skip: 3, fetch: Some(5)"));
// LocalLimitExec adjusts the `fetch`
assert!(format!("{:?}", plan).contains("LocalLimitExec"));
diff --git a/datafusion/core/tests/dataframe_functions.rs b/datafusion/core/tests/dataframe_functions.rs
index 0d3631b18..f9780345a 100644
--- a/datafusion/core/tests/dataframe_functions.rs
+++ b/datafusion/core/tests/dataframe_functions.rs
@@ -73,7 +73,7 @@ macro_rules! assert_fn_batches {
};
($EXPR:expr, $EXPECTED: expr, $LIMIT: expr) => {
let df = create_test_table()?;
- let df = df.select(vec![$EXPR])?.limit(None, Some($LIMIT))?;
+ let df = df.select(vec![$EXPR])?.limit(0, Some($LIMIT))?;
let batches = df.collect().await?;
assert_batches_eq!($EXPECTED, &batches);
diff --git a/datafusion/core/tests/sql/explain_analyze.rs b/datafusion/core/tests/sql/explain_analyze.rs
index a63839c2f..d5509cf65 100644
--- a/datafusion/core/tests/sql/explain_analyze.rs
+++ b/datafusion/core/tests/sql/explain_analyze.rs
@@ -74,7 +74,7 @@ async fn explain_analyze_baseline_metrics() {
);
assert_metrics!(
&formatted,
- "GlobalLimitExec: skip=None, fetch=3, ",
+ "GlobalLimitExec: skip=0, fetch=3, ",
"metrics=[output_rows=1, elapsed_compute="
);
assert_metrics!(
@@ -685,7 +685,7 @@ async fn test_physical_plan_display_indent() {
let physical_plan = ctx.create_physical_plan(&plan).await.unwrap();
let expected = vec![
- "GlobalLimitExec: skip=None, fetch=10",
+ "GlobalLimitExec: skip=0, fetch=10",
" SortExec: [the_min@2 DESC]",
" CoalescePartitionsExec",
" ProjectionExec: expr=[c1@0 as c1, MAX(aggregate_test_100.c12)@1 as MAX(aggregate_test_100.c12), MIN(aggregate_test_100.c12)@2 as the_min]",
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index ea923e57e..0fc80f907 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -299,7 +299,7 @@ impl LogicalPlanBuilder {
///
/// `fetch` - Maximum number of rows to fetch, after skipping `skip` rows,
/// if specified.
- pub fn limit(&self, skip: Option<usize>, fetch: Option<usize>) -> Result<Self> {
+ pub fn limit(&self, skip: usize, fetch: Option<usize>) -> Result<Self> {
Ok(Self::from(LogicalPlan::Limit(Limit {
skip,
fetch,
@@ -1057,7 +1057,7 @@ mod tests {
vec![sum(col("salary")).alias("total_salary")],
)?
.project(vec![col("state"), col("total_salary")])?
- .limit(Some(2), Some(10))?
+ .limit(2, Some(10))?
.build()?;
let expected = "Limit: skip=2, fetch=10\
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 38074f046..0cc0c28e8 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -886,7 +886,7 @@ impl LogicalPlan {
write!(
f,
"Limit: skip={}, fetch={}",
- skip.map_or("None".to_string(), |x| x.to_string()),
+ skip,
fetch.map_or_else(|| "None".to_string(), |x| x.to_string())
)
}
@@ -1294,8 +1294,9 @@ pub struct Extension {
#[derive(Clone)]
pub struct Limit {
/// Number of rows to skip before fetch
- pub skip: Option<usize>,
- /// Maximum number of rows to fetch
+ pub skip: usize,
+ /// Maximum number of rows to fetch,
+ /// None means fetching all rows
pub fetch: Option<usize>,
/// The logical plan
pub input: Arc<LogicalPlan>,
diff --git a/datafusion/optimizer/src/eliminate_limit.rs b/datafusion/optimizer/src/eliminate_limit.rs
index 6d1e1cde5..7b812869e 100644
--- a/datafusion/optimizer/src/eliminate_limit.rs
+++ b/datafusion/optimizer/src/eliminate_limit.rs
@@ -45,7 +45,7 @@ impl EliminateLimit {
/// when traversing down related to "eliminate limit".
enum Ancestor {
/// Limit
- FromLimit { skip: Option<usize> },
+ FromLimit { skip: usize },
/// Other nodes that don't affect the adjustment of "Limit"
NotRelevant,
}
@@ -65,7 +65,7 @@ fn eliminate_limit(
skip, fetch, input, ..
}) => {
let ancestor_skip = match ancestor {
- Ancestor::FromLimit { skip, .. } => skip.unwrap_or(0),
+ Ancestor::FromLimit { skip } => *skip,
_ => 0,
};
// If ancestor's skip is equal or greater than current's fetch,
@@ -82,10 +82,11 @@ fn eliminate_limit(
}
}
None => {
- match skip {
+ if *skip == 0 {
// If there is no LIMIT and OFFSET is zero, LIMIT/OFFSET can be removed
- Some(skip) if *skip == 0 => return Ok(input.as_ref().clone()),
- _ => {}
+ return Ok(input.as_ref().clone());
+ } else {
+ {}
}
}
}
@@ -174,7 +175,7 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![sum(col("b"))])
.unwrap()
- .limit(None, Some(0))
+ .limit(0, Some(0))
.unwrap()
.build()
.unwrap();
@@ -194,7 +195,7 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![sum(col("b"))])
.unwrap()
- .limit(None, Some(0))
+ .limit(0, Some(0))
.unwrap()
.union(plan1)
.unwrap()
@@ -215,9 +216,9 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![sum(col("b"))])
.unwrap()
- .limit(None, Some(2))
+ .limit(0, Some(2))
.unwrap()
- .limit(Some(2), None)
+ .limit(2, None)
.unwrap()
.build()
.unwrap();
@@ -234,11 +235,11 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![sum(col("b"))])
.unwrap()
- .limit(None, Some(2))
+ .limit(0, Some(2))
.unwrap()
.sort(vec![col("a")])
.unwrap()
- .limit(Some(2), Some(1))
+ .limit(2, Some(1))
.unwrap()
.build()
.unwrap();
@@ -255,18 +256,18 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![sum(col("b"))])
.unwrap()
- .limit(None, Some(2))
+ .limit(0, Some(2))
.unwrap()
.sort(vec![col("a")])
.unwrap()
- .limit(None, Some(1))
+ .limit(0, Some(1))
.unwrap()
.build()
.unwrap();
- let expected = "Limit: skip=None, fetch=1\
+ let expected = "Limit: skip=0, fetch=1\
\n Sort: #test.a\
- \n Limit: skip=None, fetch=2\
+ \n Limit: skip=0, fetch=2\
\n Aggregate: groupBy=[[#test.a]], aggr=[[SUM(#test.b)]]\
\n TableScan: test";
assert_optimized_plan_eq(&plan, expected);
@@ -278,11 +279,11 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![sum(col("b"))])
.unwrap()
- .limit(Some(2), Some(1))
+ .limit(2, Some(1))
.unwrap()
.sort(vec![col("a")])
.unwrap()
- .limit(Some(3), Some(1))
+ .limit(3, Some(1))
.unwrap()
.build()
.unwrap();
@@ -298,7 +299,7 @@ mod tests {
let table_scan = test_table_scan().unwrap();
let table_scan_inner = test_table_scan_with_name("test1").unwrap();
let plan = LogicalPlanBuilder::from(table_scan)
- .limit(Some(2), Some(1))
+ .limit(2, Some(1))
.unwrap()
.join_using(
&table_scan_inner,
@@ -306,7 +307,7 @@ mod tests {
vec![Column::from_name("a".to_string())],
)
.unwrap()
- .limit(Some(3), Some(1))
+ .limit(3, Some(1))
.unwrap()
.build()
.unwrap();
@@ -325,7 +326,7 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![sum(col("b"))])
.unwrap()
- .limit(Some(0), None)
+ .limit(0, None)
.unwrap()
.build()
.unwrap();
diff --git a/datafusion/optimizer/src/filter_push_down.rs b/datafusion/optimizer/src/filter_push_down.rs
index cf7b770c6..a91177b46 100644
--- a/datafusion/optimizer/src/filter_push_down.rs
+++ b/datafusion/optimizer/src/filter_push_down.rs
@@ -691,13 +691,13 @@ mod tests {
let table_scan = test_table_scan()?;
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("a"), col("b")])?
- .limit(None, Some(10))?
+ .limit(0, Some(10))?
.filter(col("a").eq(lit(1i64)))?
.build()?;
// filter is before single projection
let expected = "\
Filter: #test.a = Int64(1)\
- \n Limit: skip=None, fetch=10\
+ \n Limit: skip=0, fetch=10\
\n Projection: #test.a, #test.b\
\n TableScan: test";
assert_optimized_plan_eq(&plan, expected);
@@ -943,8 +943,8 @@ mod tests {
let table_scan = test_table_scan()?;
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("a"), col("b")])?
- .limit(None, Some(20))?
- .limit(None, Some(10))?
+ .limit(0, Some(20))?
+ .limit(0, Some(10))?
.project(vec![col("a"), col("b")])?
.filter(col("a").eq(lit(1i64)))?
.build()?;
@@ -952,8 +952,8 @@ mod tests {
let expected = "\
Projection: #test.a, #test.b\
\n Filter: #test.a = Int64(1)\
- \n Limit: skip=None, fetch=10\
- \n Limit: skip=None, fetch=20\
+ \n Limit: skip=0, fetch=10\
+ \n Limit: skip=0, fetch=20\
\n Projection: #test.a, #test.b\
\n TableScan: test";
assert_optimized_plan_eq(&plan, expected);
@@ -1030,7 +1030,7 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("a")])?
.filter(col("a").lt_eq(lit(1i64)))?
- .limit(None, Some(1))?
+ .limit(0, Some(1))?
.project(vec![col("a")])?
.filter(col("a").gt_eq(lit(1i64)))?
.build()?;
@@ -1041,7 +1041,7 @@ mod tests {
format!("{:?}", plan),
"Filter: #test.a >= Int64(1)\
\n Projection: #test.a\
- \n Limit: skip=None, fetch=1\
+ \n Limit: skip=0, fetch=1\
\n Filter: #test.a <= Int64(1)\
\n Projection: #test.a\
\n TableScan: test"
@@ -1050,7 +1050,7 @@ mod tests {
let expected = "\
Projection: #test.a\
\n Filter: #test.a >= Int64(1)\
- \n Limit: skip=None, fetch=1\
+ \n Limit: skip=0, fetch=1\
\n Projection: #test.a\
\n Filter: #test.a <= Int64(1)\
\n TableScan: test";
@@ -1064,7 +1064,7 @@ mod tests {
fn two_filters_on_same_depth() -> Result<()> {
let table_scan = test_table_scan()?;
let plan = LogicalPlanBuilder::from(table_scan)
- .limit(None, Some(1))?
+ .limit(0, Some(1))?
.filter(col("a").lt_eq(lit(1i64)))?
.filter(col("a").gt_eq(lit(1i64)))?
.project(vec![col("a")])?
@@ -1076,14 +1076,14 @@ mod tests {
"Projection: #test.a\
\n Filter: #test.a >= Int64(1)\
\n Filter: #test.a <= Int64(1)\
- \n Limit: skip=None, fetch=1\
+ \n Limit: skip=0, fetch=1\
\n TableScan: test"
);
let expected = "\
Projection: #test.a\
\n Filter: #test.a >= Int64(1) AND #test.a <= Int64(1)\
- \n Limit: skip=None, fetch=1\
+ \n Limit: skip=0, fetch=1\
\n TableScan: test";
assert_optimized_plan_eq(&plan, expected);
diff --git a/datafusion/optimizer/src/limit_push_down.rs b/datafusion/optimizer/src/limit_push_down.rs
index 66cdc144c..65bdd64ee 100644
--- a/datafusion/optimizer/src/limit_push_down.rs
+++ b/datafusion/optimizer/src/limit_push_down.rs
@@ -41,10 +41,7 @@ impl LimitPushDown {
/// when traversing down related to "limit push down".
enum Ancestor {
/// Limit
- FromLimit {
- skip: Option<usize>,
- fetch: Option<usize>,
- },
+ FromLimit { skip: usize, fetch: Option<usize> },
/// Other nodes that don't affect the adjustment of "Limit"
NotRelevant,
}
@@ -90,8 +87,7 @@ fn limit_push_down(
} => {
if let Some(fetch) = current_fetch {
// extend ancestor's fetch
- let ancestor_fetch =
- ancestor_fetch.map(|f| f + ancestor_skip.unwrap_or(0));
+ let ancestor_fetch = ancestor_fetch.map(|f| f + ancestor_skip);
let new_current_fetch =
ancestor_fetch.map_or(*fetch, |x| std::cmp::min(x, *fetch));
@@ -100,7 +96,7 @@ fn limit_push_down(
} else {
// we dont have a "fetch", and we can push down our parent's "fetch"
// extend ancestor's fetch
- ancestor_fetch.map(|f| f + ancestor_skip.unwrap_or(0))
+ ancestor_fetch.map(|f| f + ancestor_skip)
}
}
_ => *current_fetch,
@@ -139,8 +135,7 @@ fn limit_push_down(
..
},
) => {
- let ancestor_fetch =
- ancestor_skip.map_or(ancestor_fetch, |x| x + ancestor_fetch);
+ let ancestor_fetch = ancestor_fetch + ancestor_skip;
Ok(LogicalPlan::TableScan(TableScan {
table_name: table_name.clone(),
source: source.clone(),
@@ -187,18 +182,17 @@ fn limit_push_down(
},
) => {
// Push down limit through UNION
- let ancestor_fetch =
- ancestor_skip.map_or(ancestor_fetch, |x| x + ancestor_fetch);
+ let ancestor_fetch = ancestor_fetch + ancestor_skip;
let new_inputs = inputs
.iter()
.map(|x| {
Ok(Arc::new(LogicalPlan::Limit(Limit {
- skip: None,
+ skip: 0,
fetch: Some(ancestor_fetch),
input: Arc::new(limit_push_down(
_optimizer,
Ancestor::FromLimit {
- skip: None,
+ skip: 0,
fetch: Some(ancestor_fetch),
},
x,
@@ -221,8 +215,7 @@ fn limit_push_down(
..
},
) => {
- let ancestor_fetch =
- ancestor_skip.map_or(ancestor_fetch, |x| x + ancestor_fetch);
+ let ancestor_fetch = ancestor_fetch + ancestor_skip;
match join_type {
JoinType::Left => {
//if LeftOuter join push limit to left
@@ -282,7 +275,7 @@ fn generate_push_down_join(
left: Arc::new(limit_push_down(
_optimizer,
Ancestor::FromLimit {
- skip: None,
+ skip: 0,
fetch: left_limit,
},
left.as_ref(),
@@ -291,7 +284,7 @@ fn generate_push_down_join(
right: Arc::new(limit_push_down(
_optimizer,
Ancestor::FromLimit {
- skip: None,
+ skip: 0,
fetch: right_limit,
},
right.as_ref(),
@@ -370,12 +363,12 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("a")])?
- .limit(None, Some(1000))?
+ .limit(0, Some(1000))?
.build()?;
// Should push the limit down to table provider
// When it has a select
- let expected = "Limit: skip=None, fetch=1000\
+ let expected = "Limit: skip=0, fetch=1000\
\n Projection: #test.a\
\n TableScan: test, fetch=1000";
@@ -388,15 +381,15 @@ mod test {
let table_scan = test_table_scan()?;
let plan = LogicalPlanBuilder::from(table_scan)
- .limit(None, Some(1000))?
- .limit(None, Some(10))?
+ .limit(0, Some(1000))?
+ .limit(0, Some(10))?
.build()?;
// Should push down the smallest limit
// Towards table scan
// This rule doesn't replace multiple limits
- let expected = "Limit: skip=None, fetch=10\
- \n Limit: skip=None, fetch=10\
+ let expected = "Limit: skip=0, fetch=10\
+ \n Limit: skip=0, fetch=10\
\n TableScan: test, fetch=10";
assert_optimized_plan_eq(&plan, expected);
@@ -410,11 +403,11 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![max(col("b"))])?
- .limit(None, Some(1000))?
+ .limit(0, Some(1000))?
.build()?;
// Limit should *not* push down aggregate node
- let expected = "Limit: skip=None, fetch=1000\
+ let expected = "Limit: skip=0, fetch=1000\
\n Aggregate: groupBy=[[#test.a]], aggr=[[MAX(#test.b)]]\
\n TableScan: test";
@@ -429,15 +422,15 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan.clone())
.union(LogicalPlanBuilder::from(table_scan).build()?)?
- .limit(None, Some(1000))?
+ .limit(0, Some(1000))?
.build()?;
// Limit should push down through union
- let expected = "Limit: skip=None, fetch=1000\
+ let expected = "Limit: skip=0, fetch=1000\
\n Union\
- \n Limit: skip=None, fetch=1000\
+ \n Limit: skip=0, fetch=1000\
\n TableScan: test, fetch=1000\
- \n Limit: skip=None, fetch=1000\
+ \n Limit: skip=0, fetch=1000\
\n TableScan: test, fetch=1000";
assert_optimized_plan_eq(&plan, expected);
@@ -450,15 +443,15 @@ mod test {
let table_scan = test_table_scan()?;
let plan = LogicalPlanBuilder::from(table_scan)
- .limit(None, Some(1000))?
+ .limit(0, Some(1000))?
.aggregate(vec![col("a")], vec![max(col("b"))])?
- .limit(None, Some(10))?
+ .limit(0, Some(10))?
.build()?;
// Limit should use deeper LIMIT 1000, but Limit 10 shouldn't push down aggregation
- let expected = "Limit: skip=None, fetch=10\
+ let expected = "Limit: skip=0, fetch=10\
\n Aggregate: groupBy=[[#test.a]], aggr=[[MAX(#test.b)]]\
- \n Limit: skip=None, fetch=1000\
+ \n Limit: skip=0, fetch=1000\
\n TableScan: test, fetch=1000";
assert_optimized_plan_eq(&plan, expected);
@@ -470,7 +463,7 @@ mod test {
fn limit_pushdown_should_not_pushdown_limit_with_offset_only() -> Result<()> {
let table_scan = test_table_scan()?;
let plan = LogicalPlanBuilder::from(table_scan)
- .limit(Some(10), None)?
+ .limit(10, None)?
.build()?;
// Should not push any limit down to table provider
@@ -488,7 +481,7 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("a")])?
- .limit(Some(10), Some(1000))?
+ .limit(10, Some(1000))?
.build()?;
// Should push the limit down to table provider
@@ -508,12 +501,12 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("a")])?
- .limit(None, Some(1000))?
- .limit(Some(10), None)?
+ .limit(0, Some(1000))?
+ .limit(10, None)?
.build()?;
let expected = "Limit: skip=10, fetch=None\
- \n Limit: skip=None, fetch=1000\
+ \n Limit: skip=0, fetch=1000\
\n Projection: #test.a\
\n TableScan: test, fetch=1000";
@@ -528,11 +521,11 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("a")])?
- .limit(Some(10), None)?
- .limit(None, Some(1000))?
+ .limit(10, None)?
+ .limit(0, Some(1000))?
.build()?;
- let expected = "Limit: skip=None, fetch=1000\
+ let expected = "Limit: skip=0, fetch=1000\
\n Limit: skip=10, fetch=1000\
\n Projection: #test.a\
\n TableScan: test, fetch=1010";
@@ -547,16 +540,16 @@ mod test {
let table_scan = test_table_scan()?;
let plan = LogicalPlanBuilder::from(table_scan)
- .limit(Some(10), None)?
- .limit(None, Some(1000))?
- .limit(None, Some(10))?
+ .limit(10, None)?
+ .limit(0, Some(1000))?
+ .limit(0, Some(10))?
.build()?;
// Should push down the smallest limit
// Towards table scan
// This rule doesn't replace multiple limits
- let expected = "Limit: skip=None, fetch=10\
- \n Limit: skip=None, fetch=10\
+ let expected = "Limit: skip=0, fetch=10\
+ \n Limit: skip=0, fetch=10\
\n Limit: skip=10, fetch=10\
\n TableScan: test, fetch=20";
@@ -571,7 +564,7 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan)
.aggregate(vec![col("a")], vec![max(col("b"))])?
- .limit(Some(10), Some(1000))?
+ .limit(10, Some(1000))?
.build()?;
// Limit should *not* push down aggregate node
@@ -590,15 +583,15 @@ mod test {
let plan = LogicalPlanBuilder::from(table_scan.clone())
.union(LogicalPlanBuilder::from(table_scan).build()?)?
- .limit(Some(10), Some(1000))?
+ .limit(10, Some(1000))?
.build()?;
// Limit should push down through union
let expected = "Limit: skip=10, fetch=1000\
\n Union\
- \n Limit: skip=None, fetch=1010\
+ \n Limit: skip=0, fetch=1010\
\n TableScan: test, fetch=1010\
- \n Limit: skip=None, fetch=1010\
+ \n Limit: skip=0, fetch=1010\
\n TableScan: test, fetch=1010";
assert_optimized_plan_eq(&plan, expected);
@@ -618,7 +611,7 @@ mod test {
(vec!["a"], vec!["a"]),
None,
)?
- .limit(Some(10), Some(1000))?
+ .limit(10, Some(1000))?
.build()?;
// Limit pushdown Not supported in Join
@@ -644,7 +637,7 @@ mod test {
(vec!["a"], vec!["a"]),
None,
)?
- .limit(Some(10), Some(1000))?
+ .limit(10, Some(1000))?
.build()?;
// Limit pushdown Not supported in Join
@@ -671,7 +664,7 @@ mod test {
let outer_query = LogicalPlanBuilder::from(table_scan_2)
.project(vec![col("a")])?
.filter(exists(Arc::new(subquery)))?
- .limit(Some(10), Some(100))?
+ .limit(10, Some(100))?
.build()?;
// Limit pushdown Not supported in sub_query
@@ -702,7 +695,7 @@ mod test {
let outer_query = LogicalPlanBuilder::from(table_scan_2)
.project(vec![col("a")])?
.filter(exists(Arc::new(subquery)))?
- .limit(Some(10), Some(100))?
+ .limit(10, Some(100))?
.build()?;
// Limit pushdown Not supported in sub_query
@@ -732,11 +725,11 @@ mod test {
(vec!["a"], vec!["a"]),
None,
)?
- .limit(None, Some(1000))?
+ .limit(0, Some(1000))?
.build()?;
// Limit pushdown Not supported in Join
- let expected = "Limit: skip=None, fetch=1000\
+ let expected = "Limit: skip=0, fetch=1000\
\n Left Join: #test.a = #test2.a\
\n TableScan: test, fetch=1000\
\n TableScan: test2";
@@ -758,7 +751,7 @@ mod test {
(vec!["a"], vec!["a"]),
None,
)?
- .limit(Some(10), Some(1000))?
+ .limit(10, Some(1000))?
.build()?;
// Limit pushdown Not supported in Join
@@ -784,11 +777,11 @@ mod test {
(vec!["a"], vec!["a"]),
None,
)?
- .limit(None, Some(1000))?
+ .limit(0, Some(1000))?
.build()?;
// Limit pushdown Not supported in Join
- let expected = "Limit: skip=None, fetch=1000\
+ let expected = "Limit: skip=0, fetch=1000\
\n Right Join: #test.a = #test2.a\
\n TableScan: test\
\n TableScan: test2, fetch=1000";
@@ -810,7 +803,7 @@ mod test {
(vec!["a"], vec!["a"]),
None,
)?
- .limit(Some(10), Some(1000))?
+ .limit(10, Some(1000))?
.build()?;
// Limit pushdown with offset supported in right outer join
diff --git a/datafusion/optimizer/src/projection_push_down.rs b/datafusion/optimizer/src/projection_push_down.rs
index 79364eb1f..5a0b38b57 100644
--- a/datafusion/optimizer/src/projection_push_down.rs
+++ b/datafusion/optimizer/src/projection_push_down.rs
@@ -857,12 +857,12 @@ mod tests {
let plan = LogicalPlanBuilder::from(table_scan)
.project(vec![col("c"), col("a")])?
- .limit(None, Some(5))?
+ .limit(0, Some(5))?
.build()?;
assert_fields_eq(&plan, vec!["c", "a"]);
- let expected = "Limit: skip=None, fetch=5\
+ let expected = "Limit: skip=0, fetch=5\
\n Projection: #test.c, #test.a\
\n TableScan: test projection=[a, c]";
diff --git a/datafusion/optimizer/src/simplify_expressions.rs b/datafusion/optimizer/src/simplify_expressions.rs
index 1c826d7c3..d1afa3543 100644
--- a/datafusion/optimizer/src/simplify_expressions.rs
+++ b/datafusion/optimizer/src/simplify_expressions.rs
@@ -1693,7 +1693,7 @@ mod tests {
.unwrap()
.filter(col("c").not_eq(lit(false)))
.unwrap()
- .limit(None, Some(1))
+ .limit(0, Some(1))
.unwrap()
.project(vec![col("a")])
.unwrap()
@@ -1702,7 +1702,7 @@ mod tests {
let expected = "\
Projection: #test.a\
- \n Limit: skip=None, fetch=1\
+ \n Limit: skip=0, fetch=1\
\n Filter: #test.c AS test.c != Boolean(false)\
\n Filter: NOT #test.b AS test.b != Boolean(true)\
\n TableScan: test";
diff --git a/datafusion/proto/src/logical_plan.rs b/datafusion/proto/src/logical_plan.rs
index 0e1ed38cb..64a846119 100644
--- a/datafusion/proto/src/logical_plan.rs
+++ b/datafusion/proto/src/logical_plan.rs
@@ -557,11 +557,7 @@ impl AsLogicalPlan for LogicalPlanNode {
LogicalPlanType::Limit(limit) => {
let input: LogicalPlan =
into_logical_plan!(limit.input, ctx, extension_codec)?;
- let skip = if limit.skip <= 0 {
- None
- } else {
- Some(limit.skip as usize)
- };
+ let skip = limit.skip.max(0) as usize;
let fetch = if limit.fetch < 0 {
None
@@ -947,7 +943,7 @@ impl AsLogicalPlan for LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::Limit(Box::new(
protobuf::LimitNode {
input: Some(Box::new(input)),
- skip: skip.unwrap_or(0) as i64,
+ skip: *skip as i64,
fetch: fetch.unwrap_or(i64::MAX as usize) as i64,
},
))),
diff --git a/datafusion/sql/src/planner.rs b/datafusion/sql/src/planner.rs
index 5cc388ba0..edc29abf6 100644
--- a/datafusion/sql/src/planner.rs
+++ b/datafusion/sql/src/planner.rs
@@ -1310,28 +1310,25 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
}
let skip = match skip {
- Some(skip_expr) => {
- let skip = match self.sql_to_rex(
- skip_expr.value,
- input.schema(),
- &mut HashMap::new(),
- )? {
- Expr::Literal(ScalarValue::Int64(Some(s))) => {
- if s < 0 {
- return Err(DataFusionError::Plan(format!(
- "Offset must be >= 0, '{}' was provided.",
- s
- )));
- }
- Ok(s as usize)
+ Some(skip_expr) => match self.sql_to_rex(
+ skip_expr.value,
+ input.schema(),
+ &mut HashMap::new(),
+ )? {
+ Expr::Literal(ScalarValue::Int64(Some(s))) => {
+ if s < 0 {
+ return Err(DataFusionError::Plan(format!(
+ "Offset must be >= 0, '{}' was provided.",
+ s
+ )));
}
- _ => Err(DataFusionError::Plan(
- "Unexpected expression in OFFSET clause".to_string(),
- )),
- }?;
- Some(skip)
- }
- _ => None,
+ Ok(s as usize)
+ }
+ _ => Err(DataFusionError::Plan(
+ "Unexpected expression in OFFSET clause".to_string(),
+ )),
+ }?,
+ _ => 0,
};
let fetch = match fetch {
diff --git a/docs/source/user-guide/dataframe.md b/docs/source/user-guide/dataframe.md
index 4c370b6bf..e860242fc 100644
--- a/docs/source/user-guide/dataframe.md
+++ b/docs/source/user-guide/dataframe.md
@@ -42,7 +42,7 @@ let ctx = SessionContext::new();
let df = ctx.read_csv("tests/example.csv", CsvReadOptions::new()).await?;
let df = df.filter(col("a").lt_eq(col("b")))?
.aggregate(vec![col("a")], vec![min(col("b"))])?
- .limit(None, Some(100))?;
+ .limit(0, Some(100))?;
// Print results
df.show();
```