diff --git a/Cargo.lock b/Cargo.lock index 2e605c1b43eea..db5a6ebecdbe4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1878,6 +1878,7 @@ dependencies = [ "datafusion-physical-expr-adapter", "datafusion-physical-expr-common", "datafusion-physical-plan", + "datafusion-pruning", "futures", "itertools 0.15.0", "log", diff --git a/datafusion/catalog-listing/Cargo.toml b/datafusion/catalog-listing/Cargo.toml index abe58f45994be..62f7d78d118b5 100644 --- a/datafusion/catalog-listing/Cargo.toml +++ b/datafusion/catalog-listing/Cargo.toml @@ -42,6 +42,7 @@ datafusion-physical-expr = { workspace = true } datafusion-physical-expr-adapter = { workspace = true } datafusion-physical-expr-common = { workspace = true } datafusion-physical-plan = { workspace = true } +datafusion-pruning = { workspace = true } futures = { workspace = true } itertools = { workspace = true } log = { workspace = true } diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index c94ddcef22d6b..89f4840a34baa 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/datafusion/catalog-listing/src/table.rs @@ -45,15 +45,19 @@ use datafusion_execution::cache::cache_manager::{ use datafusion_expr::dml::InsertOp; use datafusion_expr::execution_props::ExecutionProps; use datafusion_expr::physical_planning_context::PhysicalPlanningContext; +use datafusion_expr::utils::conjunction; use datafusion_expr::{ Expr, Partitioning as LogicalPartitioning, TableProviderFilterPushDown, TableType, }; -use datafusion_physical_expr::{create_lex_ordering, create_physical_partitioning}; +use datafusion_physical_expr::{ + Partitioning, create_lex_ordering, create_physical_expr, create_physical_partitioning, +}; use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; use datafusion_physical_expr_common::sort_expr::LexOrdering; use datafusion_physical_plan::ExecutionPlan; use datafusion_physical_plan::empty::EmptyExec; use datafusion_physical_plan::limit::GlobalLimitExec; +use datafusion_pruning::{PruningPredicateBuilder, RangePruningStatistics}; use futures::future::BoxFuture; use futures::{Stream, StreamExt, TryStreamExt, future, stream}; use object_store::ObjectStore; @@ -647,10 +651,17 @@ impl ListingTable { // If the filters can be resolved using only partition cols, there is no need to // pushdown it to TableScan, otherwise, `unhandled` pruning predicates will be generated - let (partition_filters, filters): (Vec<_>, Vec<_>) = - filters.iter().cloned().partition(|filter| { - can_be_evaluated_for_partition_pruning(&table_partition_col_names, filter) - }); + let scan_filters = filters; + let filters = scan_filters + .iter() + .filter(|filter| { + !can_be_evaluated_for_partition_pruning( + &table_partition_col_names, + filter, + ) + }) + .cloned() + .collect::>(); let declared_output_partitioning = self.options.output_partitioning.as_ref(); @@ -672,7 +683,7 @@ impl ListingTable { grouped_by_partition: partitioned_by_file_group, partition_split_points, } = self - .list_files_for_scan(state, &partition_filters, statistic_file_limit) + .list_files_for_scan(state, &scan_filters, statistic_file_limit) .await?; // if no files need to be read, return an `EmptyExec` @@ -918,22 +929,41 @@ impl ListingTable { /// be distributed to different threads / executors. /// /// If [`ListingOptions::output_partitioning`] is set, returns one file - /// group per declared partition, including empty trailing groups. + /// group per declared partition, including empty trailing groups. Filters + /// known not to match declared ranges leave empty groups at those indices. pub async fn list_files_for_scan<'a>( &'a self, ctx: &'a dyn Session, filters: &'a [Expr], limit: Option, ) -> datafusion_common::Result { + let partition_column_names = self + .options + .table_partition_cols + .iter() + .map(|(name, _)| name.as_str()) + .collect::>(); + let partition_filters = filters + .iter() + .filter(|filter| { + can_be_evaluated_for_partition_pruning(&partition_column_names, filter) + }) + .cloned() + .collect::>(); if let Some(output_partitioning) = self.options.output_partitioning.as_ref() { self.list_files_for_declared_output_partitioning( ctx, output_partitioning, + &partition_filters, filters, ) .await } else { - self.list_files_for_regular_scan(ctx, filters, limit).await + let file_limit = (partition_filters.len() == filters.len()) + .then_some(limit) + .flatten(); + self.list_files_for_regular_scan(ctx, &partition_filters, file_limit) + .await } } @@ -1056,6 +1086,7 @@ impl ListingTable { ctx: &'a dyn Session, output_partitioning: &LogicalPartitioning, filters: &'a [Expr], + range_filters: &[Expr], ) -> datafusion_common::Result { let Some(file_group_count) = output_partitioning.partition_count() else { return datafusion_common::not_impl_err!( @@ -1084,12 +1115,69 @@ impl ListingTable { if !file_groups.is_empty() { file_groups.resize_with(file_group_count, || FileGroup::new(vec![])); } - let file_groups = + let mut file_groups = self.filter_declared_file_groups_by_partition_filters(file_groups, filters)?; + if matches!(output_partitioning, LogicalPartitioning::Range(_)) { + self.prune_range_file_groups( + ctx, + output_partitioning, + &mut file_groups, + range_filters, + )?; + } + self.list_files_result_from_groups(ctx, file_groups, inexact_stats, false, vec![]) } + fn prune_range_file_groups( + &self, + ctx: &dyn Session, + output_partitioning: &LogicalPartitioning, + file_groups: &mut [FileGroup], + filters: &[Expr], + ) -> datafusion_common::Result<()> { + // Do not invoke volatile expressions while planning. Unsupported + // expressions and pruning evaluation errors must keep the partitions. + let Some(filter) = + conjunction(filters.iter().filter(|f| !f.is_volatile()).cloned()) + else { + return Ok(()); + }; + let schema = DFSchema::try_from(Arc::clone(&self.table_schema))?; + let planning_ctx = PhysicalPlanningContext::default(); + let Partitioning::Range(range) = create_physical_partitioning( + output_partitioning, + &schema, + ctx.execution_props(), + &planning_ctx, + )? + else { + return Ok(()); + }; + let retained = + create_physical_expr(&filter, &schema, ctx.execution_props(), &planning_ctx) + .and_then(|expr| { + PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&self.table_schema)) + .try_build(expr) + }) + .and_then(|predicate| { + predicate.prune(&RangePruningStatistics::new(&range)) + }); + match retained { + Ok(retained) => { + for (group, keep) in file_groups.iter_mut().zip(retained) { + if !keep { + *group = FileGroup::new(vec![]); + } + } + } + Err(error) => log::debug!("Skipping range partition pruning: {error}"), + } + Ok(()) + } + fn filter_declared_file_groups_by_partition_filters( &self, file_groups: Vec, diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs index c10ae42477f4d..8130a4caa6292 100644 --- a/datafusion/core/src/datasource/listing/table.rs +++ b/datafusion/core/src/datasource/listing/table.rs @@ -1431,6 +1431,75 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_list_files_prunes_normalized_range_boundaries() -> Result<()> { + let files = [ + "bucket/key-prefix/file0", + "bucket/key-prefix/file1", + "bucket/key-prefix/file2", + "bucket/key-prefix/file3", + ]; + let ctx = SessionContext::new(); + // Logical split points need not have the physical range column's type. + let opt = ListingOptions::new(Arc::new(JsonFormat::default())) + .with_file_extension_opt(Some("")) + .with_output_partitioning(Some(LogicalPartitioning::Range( + LogicalRangePartitioning::try_new( + vec![col("a").sort(true, true)], + [10_i64, 20, 30] + .into_iter() + .map(|value| SplitPoint::new(vec![ScalarValue::from(value)])) + .collect(), + )?, + ))); + let table = listing_table_with_files( + &ctx, + &files, + "test:///bucket/key-prefix/", + opt, + Schema::new(vec![Field::new("a", DataType::Int32, true)]), + )?; + + for (filter, expected) in [ + (col("a").lt(lit(10_i32)), vec![1, 0, 0, 0]), + (col("a").eq(lit(10_i32)), vec![0, 1, 0, 0]), + (col("a").gt_eq(lit(30_i32)), vec![0, 0, 0, 1]), + (col("a").is_null(), vec![1, 0, 0, 0]), + ] { + let listed = table + .list_files_for_scan(&ctx.state(), std::slice::from_ref(&filter), Some(1)) + .await?; + assert!(!listed.grouped_by_partition); + assert!(listed.partition_split_points.is_empty()); + assert_eq!( + listed + .file_groups + .iter() + .map(|g| g.len()) + .collect::>(), + expected, + "{filter}" + ); + let scan = table.scan(&ctx.state(), None, &[filter], None).await?; + assert_eq!(scan.output_partitioning().partition_count(), 4); + } + // Invalid pruning expressions are conservative; the normal scan/filter + // planning remains responsible for reporting any expression error. + let invalid_filter = col("missing").eq(lit(10_i32)); + let listed = table + .list_files_for_scan(&ctx.state(), &[invalid_filter], None) + .await?; + assert_eq!( + listed + .file_groups + .iter() + .map(|g| g.len()) + .collect::>(), + vec![1, 1, 1, 1] + ); + Ok(()) + } + #[tokio::test] async fn test_range_output_partitioning_normalizes_split_point_types() -> Result<()> { let files = ["bucket/key-prefix/file0"]; diff --git a/datafusion/pruning/src/lib.rs b/datafusion/pruning/src/lib.rs index 3e7c2ad7aa7ab..f587c8b145f2e 100644 --- a/datafusion/pruning/src/lib.rs +++ b/datafusion/pruning/src/lib.rs @@ -22,6 +22,7 @@ mod in_list; mod key_range_bitmap_expr; mod primitive_in_list; mod pruning_predicate; +mod range; mod string_in_list; pub use file_pruner::FilePruner; @@ -29,3 +30,4 @@ pub use pruning_predicate::{ MAX_IN_LIST_SIZE, PredicateRewriter, PruningPredicate, PruningPredicateBuilder, PruningStatistics, RequiredColumns, UnhandledPredicateHook, build_pruning_predicate, }; +pub use range::RangePruningStatistics; diff --git a/datafusion/pruning/src/range.rs b/datafusion/pruning/src/range.rs new file mode 100644 index 0000000000000..2d2e4d82b327a --- /dev/null +++ b/datafusion/pruning/src/range.rs @@ -0,0 +1,419 @@ +// 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. + +use std::collections::HashSet; +use std::sync::Arc; + +use arrow::array::{ArrayRef, BooleanArray, UInt64Array}; +use datafusion_common::{Column, ScalarValue}; +use datafusion_physical_expr::RangePartitioning; +use datafusion_physical_expr::expressions::Column as PhysicalColumn; + +use crate::PruningStatistics; + +/// Bounds implied by the effective split points of a [`RangePartitioning`]. +/// +/// Each partition is a container for [`crate::PruningPredicate`]. The returned +/// mask can be used by scans or physical operators without changing partition +/// identities. The range metadata must describe the rows' actual distribution. +/// +/// Currently, min/max bounds are provided for a single direct integer column. +/// Other keys are unknown, including lexicographic multi-column ranges: applying +/// each component's split points independently would not give valid bounds. +/// Exclusive integer endpoints are converted to closed bounds with checked +/// arithmetic. Overflow leaves the bound unknown rather than wrapping it. +/// +/// # Public Only for Internal Use: +/// `datafusion-catalog-listing` uses this adapter to prune declared file groups. +/// It can also be shared by physical optimizers without depending on catalog +/// implementations; it is not yet a supported downstream extension point. +#[doc(hidden)] +pub struct RangePruningStatistics<'a> { + range: &'a RangePartitioning, +} + +impl<'a> RangePruningStatistics<'a> { + /// Create statistics from normalized physical range metadata. + pub fn new(range: &'a RangePartitioning) -> Self { + Self { range } + } + + fn matches_column(&self, column: &Column) -> bool { + let [sort] = self.range.ordering().as_ref() else { + return false; + }; + sort.expr + .downcast_ref::() + .is_some_and(|key| key.name() == column.name()) + } + + fn bounds(&self, column: &Column, minimum: bool) -> Option { + if !self.matches_column(column) { + return None; + } + let data_type = self.range.split_points().first()?.values()[0].data_type(); + if !data_type.is_integer() { + return None; + } + let unknown = ScalarValue::try_from(&data_type).ok()?; + let one = ScalarValue::new_one(&data_type).ok()?; + let descending = self.range.ordering()[0].options.descending; + // In sort order, the preceding split is inclusive and the following + // split exclusive. DESC reverses their roles as numeric min/max. + let exclusive = minimum == descending; + ScalarValue::iter_to_array((0..self.num_containers()).map(|index| { + let split = if exclusive { + self.range.split_points().get(index) + } else { + index + .checked_sub(1) + .and_then(|i| self.range.split_points().get(i)) + }; + let Some(value) = split.map(|point| &point.values()[0]) else { + return unknown.clone(); + }; + if value.is_null() { + return unknown.clone(); + } + if exclusive { + if minimum { + value.add_checked(&one) + } else { + value.sub_checked(&one) + } + .unwrap_or_else(|_| unknown.clone()) + } else { + value.clone() + } + })) + .ok() + } +} + +impl PruningStatistics for RangePruningStatistics<'_> { + fn min_values(&self, column: &Column) -> Option { + self.bounds(column, true) + } + + fn max_values(&self, column: &Column) -> Option { + self.bounds(column, false) + } + + fn num_containers(&self) -> usize { + self.range.partition_count() + } + + fn null_counts(&self, column: &Column) -> Option { + if !self.matches_column(column) { + return None; + } + // NULL compares equal to a NULL split and belongs after that split. + // Without such a split it belongs in the first/last sort-order bucket. + let null_partition = if self.range.ordering()[0].options.nulls_first { + self.range + .split_points() + .iter() + .take_while(|point| point.values()[0].is_null()) + .count() + } else { + self.num_containers() - 1 + }; + Some(Arc::new(UInt64Array::from_iter( + (0..self.num_containers()) + .map(|index| (index != null_partition).then_some(0)), + ))) + } + + fn row_counts(&self) -> Option { + None + } + + fn contained( + &self, + _column: &Column, + _values: &HashSet, + ) -> Option { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::cmp::Ordering; + + use arrow::compute::SortOptions; + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_common::{DFSchema, Result, SplitPoint}; + use datafusion_expr::execution_props::ExecutionProps; + use datafusion_expr::physical_planning_context::PhysicalPlanningContext; + use datafusion_expr::{Expr, col, lit}; + use datafusion_physical_expr::create_physical_expr; + use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; + + use crate::PruningPredicateBuilder; + + fn range( + values: Vec, + descending: bool, + nulls_first: bool, + ) -> RangePartitioning { + RangePartitioning::try_new( + LexOrdering::new(vec![PhysicalSortExpr::new( + Arc::new(PhysicalColumn::new("k", 0)), + SortOptions { + descending, + nulls_first, + }, + )]) + .unwrap(), + values + .into_iter() + .map(|value| SplitPoint::new(vec![value])) + .collect(), + ) + .unwrap() + } + + fn prune(range: &RangePartitioning, expr: Expr, data_type: DataType) -> Vec { + let schema = Arc::new(Schema::new(vec![ + Field::new("k", data_type, true), + Field::new("other", DataType::Int32, true), + ])); + let expr = create_physical_expr( + &expr, + &DFSchema::try_from(Arc::clone(&schema)).unwrap(), + &ExecutionProps::new(), + &PhysicalPlanningContext::default(), + ) + .unwrap(); + PruningPredicateBuilder::new() + .with_file_schema(schema) + .try_build(expr) + .unwrap() + .prune(&RangePruningStatistics::new(range)) + .unwrap() + } + + #[test] + fn integer_boundaries_and_compound_predicates() { + let ascending = range( + [10, 20, 30].into_iter().map(ScalarValue::from).collect(), + false, + false, + ); + for (expr, expected) in [ + (col("k").lt(lit(10_i32)), vec![true, false, false, false]), + (col("k").gt_eq(lit(30_i32)), vec![false, false, false, true]), + (col("k").eq(lit(20_i32)), vec![false, false, true, false]), + ( + col("k").gt_eq(lit(10_i32)).and(col("k").lt(lit(20_i32))), + vec![false, true, false, false], + ), + ( + col("k").lt(lit(10_i32)).or(col("k").gt_eq(lit(30_i32))), + vec![true, false, false, true], + ), + (col("other").eq(lit(5_i32)), vec![true; 4]), + ] { + assert_eq!(prune(&ascending, expr, DataType::Int32), expected); + } + let descending = range( + [30, 20, 10].into_iter().map(ScalarValue::from).collect(), + true, + false, + ); + assert_eq!( + prune(&descending, col("k").eq(lit(20_i32)), DataType::Int32), + [false, false, true, false] + ); + assert_eq!( + prune(&descending, col("k").lt(lit(10_i32)), DataType::Int32), + [false, false, false, true] + ); + assert_eq!( + prune(&descending, col("k").gt(lit(30_i32)), DataType::Int32), + [true, false, false, false] + ); + } + + #[test] + fn all_int8_values_and_nulls_are_conservative() -> Result<()> { + for descending in [false, true] { + for nulls_first in [false, true] { + let mut values = if descending { + vec![127_i8, 0, -128] + } else { + vec![-128_i8, 0, 127] + } + .into_iter() + .map(ScalarValue::from) + .collect::>(); + if nulls_first { + values.insert(0, ScalarValue::Int8(None)); + } else { + values.push(ScalarValue::Int8(None)); + } + let range = range(values, descending, nulls_first); + let compare = |row: Option, point: &ScalarValue| { + let ScalarValue::Int8(point) = point else { + unreachable!() + }; + match (row, point) { + (None, None) => Ordering::Equal, + (None, _) => { + if nulls_first { + Ordering::Less + } else { + Ordering::Greater + } + } + (_, None) => { + if nulls_first { + Ordering::Greater + } else { + Ordering::Less + } + } + (Some(row), Some(point)) => { + if descending { + point.cmp(&row) + } else { + row.cmp(point) + } + } + } + }; + for threshold in [-128_i8, -1, 0, 1, 127] { + for (expr, matches) in [ + ( + col("k").lt(lit(threshold)), + (|v, t| v < t) as fn(i8, i8) -> bool, + ), + ( + col("k").gt_eq(lit(threshold)), + (|v, t| v >= t) as fn(i8, i8) -> bool, + ), + ( + col("k").eq(lit(threshold)), + (|v, t| v == t) as fn(i8, i8) -> bool, + ), + ] { + let retained = prune(&range, expr, DataType::Int8); + for row in i8::MIN..=i8::MAX { + if matches(row, threshold) { + let index = + range.split_points().partition_point(|point| { + compare(Some(row), &point.values()[0]) + != Ordering::Less + }); + assert!( + retained[index], + "lost {row}, threshold {threshold}, DESC={descending}, NULLS FIRST={nulls_first}" + ); + } + } + } + } + let retained = prune(&range, col("k").is_null(), DataType::Int8); + let null_index = range.split_points().partition_point(|point| { + compare(None, &point.values()[0]) != Ordering::Less + }); + assert!(retained[null_index]); + assert_eq!(retained.iter().filter(|keep| **keep).count(), 1); + } + } + Ok(()) + } + + #[test] + fn sampled_layout_uses_effective_boundaries() { + let sampled = range( + [10, 20, 30, 40, 50] + .into_iter() + .map(ScalarValue::from) + .collect(), + false, + false, + ); + let scaled = sampled.scale(3).unwrap(); + assert_eq!( + prune(&scaled, col("k").lt(lit(20_i32)), DataType::Int32), + [true, false, false] + ); + } + + #[test] + fn unsupported_keys_and_empty_boundaries_stay_unknown() { + let empty = range(vec![], false, false); + assert_eq!( + prune(&empty, col("k").lt(lit(20_i32)), DataType::Int32), + [true] + ); + + let strings = range(vec![ScalarValue::from("m")], false, false); + assert_eq!( + prune(&strings, col("k").eq(lit("z")), DataType::Utf8), + [true, true] + ); + + let multi_key = RangePartitioning::try_new( + LexOrdering::new(vec![ + PhysicalSortExpr::new( + Arc::new(PhysicalColumn::new("k", 0)), + SortOptions::default(), + ), + PhysicalSortExpr::new( + Arc::new(PhysicalColumn::new("other", 1)), + SortOptions::default(), + ), + ]) + .unwrap(), + vec![SplitPoint::new(vec![ + ScalarValue::from(10_i32), + ScalarValue::from(20_i32), + ])], + ) + .unwrap(); + assert_eq!( + prune(&multi_key, col("k").eq(lit(10_i32)), DataType::Int32), + [true, true] + ); + } + + #[test] + fn unsigned_endpoints_do_not_wrap() { + let ascending = range( + vec![ScalarValue::from(0_u8), ScalarValue::from(255_u8)], + false, + false, + ); + assert_eq!( + prune(&ascending, col("k").gt_eq(lit(255_u8)), DataType::UInt8), + [true, false, true] + ); + let descending = range( + vec![ScalarValue::from(255_u8), ScalarValue::from(0_u8)], + true, + false, + ); + assert_eq!( + prune(&descending, col("k").lt_eq(lit(0_u8)), DataType::UInt8), + [true, false, true] + ); + } +} diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index 21f9f439a2085..4bfe89cb9f9d1 100644 --- a/datafusion/sqllogictest/test_files/range_partitioning.slt +++ b/datafusion/sqllogictest/test_files/range_partitioning.slt @@ -41,6 +41,45 @@ set datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = false; statement ok set datafusion.execution.parquet.pushdown_filters = false; +# A filter below the first split point should leave the other file groups empty +# while retaining all four range partition indexes. +query TT +EXPLAIN SELECT range_key FROM range_partitioned WHERE range_key < 10; +---- +physical_plan +01)FilterExec: range_key@0 < 10 +02)--DataSourceExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.parquet], [], [], []]}, projection=[range_key], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=range_key@0 < 10, pruning_predicate=range_key_null_count@1 != row_count@2 AND range_key_min@0 < 10, required_guarantees=[] + +# A row equal to a split point belongs to the partition after that split. +query TT +EXPLAIN SELECT range_key FROM range_partitioned WHERE range_key = 10; +---- +physical_plan +01)FilterExec: range_key@0 = 10 +02)--DataSourceExec: file_groups={4 groups: [[], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.parquet], [], []]}, projection=[range_key], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=range_key@0 = 10, pruning_predicate=range_key_null_count@2 != row_count@3 AND range_key_min@0 <= 10 AND 10 <= range_key_max@1, required_guarantees=[range_key in (10)] + +# Pruning must preserve LIMIT/OFFSET results after the other groups become empty. +query I +SELECT range_key FROM range_partitioned +WHERE range_key < 10 ORDER BY range_key LIMIT 1 OFFSET 1; +---- +5 + +query I +SELECT range_key FROM range_partitioned +WHERE range_key < 10 ORDER BY range_key LIMIT 1 OFFSET 100; +---- + +# A range-key join must still align the surviving rows with their partition. +query II +SELECT l.range_key, r.value +FROM (SELECT range_key FROM range_partitioned WHERE range_key < 10) l +JOIN range_partitioned r ON l.range_key = r.range_key +ORDER BY l.range_key; +---- +1 10 +5 50 + ########## # TEST 1: Aggregate on Range Partition Column # With subset threshold met and preserve-file disabled, Range([range_key])