diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs index 2dbacf1d898ac..489076331bbf5 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs @@ -1030,6 +1030,104 @@ fn range_right_mark_hash_join_reuses_range_partitioning() -> Result<()> { Ok(()) } +#[test] +fn range_hash_join_repartitions_unsatisfied_side_to_match_range() -> Result<()> { + let left = parquet_exec_with_output_partitioning(range_partitioning( + "b", + [100, 200, 300], + SortOptions::default(), + )?); + let right = parquet_exec_with_output_partitioning(range_partitioning( + "a", + [10, 20, 30], + SortOptions::default(), + )?); + let join_on = vec![( + Arc::new(Column::new_with_schema("a", &left.schema())?) as _, + Arc::new(Column::new_with_schema("a", &right.schema())?) as _, + )]; + let join = hash_join_exec(left, right, &join_on, &JoinType::Inner); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + assert_plan!( + plan, + @r" + HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0)] + RepartitionExec: partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), input_partitions=4 + DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([b@1 ASC], [(100), (200), (300)], 4), file_type=parquet + DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet + " + ); + + Ok(()) +} + +#[test] +fn range_hash_join_repartitions_unpartitioned_side_to_match_range() -> Result<()> { + let left = parquet_exec(); + let right = parquet_exec_with_output_partitioning(range_partitioning( + "a", + [10, 20, 30], + SortOptions::default(), + )?); + let join_on = vec![( + Arc::new(Column::new_with_schema("a", &left.schema())?) as _, + Arc::new(Column::new_with_schema("a", &right.schema())?) as _, + )]; + let join = hash_join_exec(left, right, &join_on, &JoinType::Inner); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + assert_plan!( + plan, + @r" + HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0)] + RepartitionExec: partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), input_partitions=1 + DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet + DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet + " + ); + + Ok(()) +} + +#[test] +fn range_hash_join_rehashes_incompatible_data_type() -> Result<()> { + let left = parquet_exec(); + let right = parquet_exec_with_output_partitioning(range_partitioning( + "a", + [10, 20, 30], + SortOptions::default(), + )?); + let join_on = vec![( + Arc::new(Column::new_with_schema("d", &left.schema())?) as _, + Arc::new(Column::new_with_schema("a", &right.schema())?) as _, + )]; + let join = hash_join_exec(left, right, &join_on, &JoinType::Inner); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + assert_plan!( + plan, + @r" + HashJoinExec: mode=Partitioned, join_type=Inner, on=[(d@3, a@0)] + RepartitionExec: partitioning=Hash([d@3], 4), input_partitions=1 + DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet + RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4 + DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet + " + ); + + Ok(()) +} + #[test] fn range_right_semi_hash_join_rehashes_incompatible_sort_options() -> Result<()> { let left = parquet_exec_with_output_partitioning(range_partitioning( diff --git a/datafusion/physical-expr/src/partitioning.rs b/datafusion/physical-expr/src/partitioning.rs index 98f082f7256db..ff3d364aabb50 100644 --- a/datafusion/physical-expr/src/partitioning.rs +++ b/datafusion/physical-expr/src/partitioning.rs @@ -21,6 +21,7 @@ use crate::{ EquivalenceProperties, PhysicalExpr, equivalence::ProjectionMapping, expressions::UnKnownColumn, physical_exprs_contains, physical_exprs_equal, }; +use arrow::datatypes::Schema; pub use datafusion_common::SplitPoint; use datafusion_common::{Result, validate_range_split_points}; use datafusion_physical_expr_common::physical_expr::format_physical_expr_list; @@ -281,6 +282,49 @@ impl RangePartitioning { split_points: self.split_points.clone(), }) } + + /// Checks whether the types of the given expressions match the data types of the split points in this range partitioning. + pub fn is_compatible_with_expressions( + &self, + exprs: &[Arc], + schema: &Schema, + ) -> bool { + if self.ordering.len() != exprs.len() { + return false; + } + if let Some(first_split) = self.split_points.first() { + exprs.iter().zip(first_split.values()).all(|(expr, val)| { + expr.data_type(schema) + .map(|dt| dt == val.data_type()) + .unwrap_or(false) + }) + } else { + true + } + } + + /// Adapts this range partitioning to the given expressions, preserving split points and sort options. + /// Returns `None` if `exprs` count doesn't match ordering length or expression types don't match split points. + pub fn adapt( + &self, + exprs: &[Arc], + schema: &Schema, + ) -> Option { + if !self.is_compatible_with_expressions(exprs, schema) { + return None; + } + let new_ordering = LexOrdering::new( + exprs + .iter() + .zip(&self.ordering) + .map(|(expr, sort_expr)| PhysicalSortExpr { + expr: Arc::clone(expr), + options: sort_expr.options, + }) + .collect::>(), + )?; + Self::try_new(new_ordering, self.split_points.clone()).ok() + } } impl Display for RangePartitioning { @@ -517,6 +561,37 @@ impl Partitioning { } } } + + /// Adapts this partitioning scheme to satisfy a required [`Distribution`] on the given schema. + /// + /// - For `Partitioning::Hash`: creates `Partitioning::Hash(exprs, partition_count)`. + /// - For `Partitioning::Range`: adapts the range partitioning to the requirement's expressions using [`RangePartitioning::adapt`]. + /// - For other partitioning schemes: returns `None`. + #[expect( + deprecated, + reason = "HashPartitioned is accepted during the KeyPartitioned migration" + )] + pub fn adapt( + &self, + child_requirement: &Distribution, + child_schema: &Schema, + ) -> Option { + let (Distribution::HashPartitioned(exprs) | Distribution::KeyPartitioned(exprs)) = + child_requirement + else { + return None; + }; + + match self { + Partitioning::Range(ref_range) => ref_range + .adapt(exprs, child_schema) + .map(Partitioning::Range), + Partitioning::Hash(_, ref_count) => { + Some(Partitioning::Hash(exprs.to_vec(), *ref_count)) + } + _ => None, + } + } } /// Protobuf conversions for [`Partitioning`]. @@ -1291,6 +1366,117 @@ mod tests { Ok(()) } + + #[test] + fn test_range_partitioning_adapt() -> Result<()> { + let fixture = PartitioningTestFixture::new(vec![ + ("a", DataType::Int32), + ("b", DataType::Int64), + ("c", DataType::Int32), + ])?; + + let range = fixture.range( + [0], + vec![ + SplitPoint::new(vec![ScalarValue::Int32(Some(10))]), + SplitPoint::new(vec![ScalarValue::Int32(Some(20))]), + ], + ); + + // Adapting to col_c (same type Int32) succeeds + let adapted = range.adapt(&[fixture.col(2)], &fixture.schema).unwrap(); + assert_eq!(adapted.ordering().len(), 1); + assert!(adapted.ordering()[0].expr.eq(&fixture.col(2))); + assert_eq!(adapted.partition_count(), 3); + + // Adapting to col_b (different type Int64) fails + assert!(range.adapt(&[fixture.col(1)], &fixture.schema).is_none()); + + // Adapting to empty or mismatch count fails + assert!(range.adapt(&[], &fixture.schema).is_none()); + assert!( + range + .adapt(&fixture.cols([0, 2]), &fixture.schema) + .is_none() + ); + + // Partitioning::adapt works with Distribution::KeyPartitioned + let part = Partitioning::Range(range); + assert!( + part.adapt(&fixture.key_distribution([1]), &fixture.schema) + .is_none() + ); + + let adapted_part = part + .adapt(&fixture.key_distribution([2]), &fixture.schema) + .unwrap(); + match adapted_part { + Partitioning::Range(r) => assert!(r.ordering()[0].expr.eq(&fixture.col(2))), + _ => panic!("expected Range partitioning"), + } + + // Partitioning::Hash adaptation + let hash_part = fixture.hash_partitioning([1], 4); + let adapted_hash = hash_part + .adapt(&fixture.key_distribution([2]), &fixture.schema) + .unwrap(); + match adapted_hash { + Partitioning::Hash(exprs, count) => { + assert_eq!(count, 4); + assert_eq!(exprs.len(), 1); + assert!(exprs[0].eq(&fixture.col(2))); + } + _ => panic!("expected Hash partitioning"), + } + + Ok(()) + } + + #[test] + fn test_range_partitioning_adapt_multi_key() -> Result<()> { + let fixture = PartitioningTestFixture::new(vec![ + ("k1", DataType::Int32), + ("k2", DataType::Utf8), + ("t1", DataType::Int32), + ("t2", DataType::Utf8), + ])?; + + let opt_k1 = SortOptions { + descending: true, + nulls_first: false, + }; + let opt_k2 = SortOptions { + descending: false, + nulls_first: true, + }; + + let ordering = LexOrdering::new(vec![ + fixture.range_sort_expr(0, opt_k1), + fixture.range_sort_expr(1, opt_k2), + ]) + .unwrap(); + + let split_points = vec![ + SplitPoint::new(vec![ScalarValue::Int32(Some(20)), ScalarValue::Utf8(None)]), + SplitPoint::new(vec![ + ScalarValue::Int32(Some(10)), + ScalarValue::Utf8(Some("foo".to_string())), + ]), + ]; + + let range = RangePartitioning::try_new(ordering, split_points.clone())?; + let adapted = range.adapt(&fixture.cols([2, 3]), &fixture.schema).unwrap(); + + assert_eq!(adapted.ordering().len(), 2); + assert!(adapted.ordering()[0].expr.eq(&fixture.col(2))); + assert_eq!(adapted.ordering()[0].options, opt_k1); + assert!(adapted.ordering()[1].expr.eq(&fixture.col(3))); + assert_eq!(adapted.ordering()[1].options, opt_k2); + assert_eq!(adapted.split_points(), &split_points); + assert_eq!(adapted.partition_count(), 3); + + Ok(()) + } } #[cfg(all(test, feature = "proto"))] diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs index 07bc98b2db798..0368577f9a24f 100644 --- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs +++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs @@ -1039,14 +1039,67 @@ fn get_repartition_requirement_status( .collect()) } +/// Helper to rank execution plans by estimated data size (bytes, then rows) +/// when selecting a reference candidate for co-partitioning. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +struct PlanSize { + byte_size: Option, + num_rows: Option, +} + +impl PlanSize { + fn from_plan(plan: &dyn ExecutionPlan) -> Self { + let stats = StatisticsContext::new() + .compute(plan, &StatisticsArgs::new()) + .ok(); + Self { + byte_size: stats + .as_ref() + .and_then(|s| s.total_byte_size.get_value().copied()), + num_rows: stats.as_ref().and_then(|s| s.num_rows.get_value().copied()), + } + } + + fn is_known(&self) -> bool { + self.byte_size.is_some() || self.num_rows.is_some() + } +} + +impl PartialOrd for PlanSize { + fn partial_cmp(&self, other: &Self) -> Option { + // 1. Compare byte_size if present on both sides + if let (Some(a), Some(b)) = (self.byte_size, other.byte_size) { + match a.cmp(&b) { + std::cmp::Ordering::Equal => {} + ord => return Some(ord), + } + } + + // 2. Compare num_rows if present on both sides + if let (Some(a), Some(b)) = (self.num_rows, other.num_rows) { + match a.cmp(&b) { + std::cmp::Ordering::Equal => {} + ord => return Some(ord), + } + } + + // 3. If any metric was present and equal on both, return Equal + if (self.byte_size.is_some() && self.byte_size == other.byte_size) + || (self.num_rows.is_some() && self.num_rows == other.num_rows) + { + return Some(std::cmp::Ordering::Equal); + } + + // Incomparable (no metric present on both sides) + None + } +} + /// Enforce cross-child distribution relationships after each child has already /// satisfied its own distribution requirement. /// /// See [`InputDistributionRequirements`] for the distinction between /// independent per-child requirements and co-partitioned child relationships. -/// -/// Currently, unsatisfied co-partitioning is repaired by hash repartitioning -/// key-partitioned children and other relationship kinds are rejected. #[expect( deprecated, reason = "HashPartitioned is accepted during the KeyPartitioned migration" @@ -1071,46 +1124,179 @@ fn enforce_distribution_relationships( return Ok(()); } + // Check if any co-partitioned child already satisfies its distribution requirement + // and can serve as a reference partitioning. + let mut satisfied_children = Vec::new(); + for &i in &unsatisfied_children { + let child = &children[i]; + let plan = child.context.plan.as_ref(); + if let Ok(satisfaction) = input_distributions.child_satisfaction( + i, + plan, + ChildSatisfactionOptions::new(), + ) && satisfaction.is_satisfied() + { + let partitioning = plan.output_partitioning(); + match partitioning { + Partitioning::Range(_) | Partitioning::Hash(_, _) => { + let is_native = !plan.is::(); + satisfied_children.push((i, partitioning.clone(), is_native)); + } + _ => {} + } + } + } + + let best_satisfied_child: Option<(usize, Partitioning)> = match satisfied_children + .len() + { + 0 => None, + 1 => satisfied_children + .into_iter() + .next() + .map(|(i, p, _)| (i, p)), + _ => { + // Prefer native partitioned children over newly repartitioned ones + let native_children: Vec<_> = satisfied_children + .iter() + .filter(|(_, _, is_native)| *is_native) + .collect(); + if native_children.len() == 1 { + let (i, p, _) = native_children[0]; + Some((*i, p.clone())) + } else { + let pool = if !native_children.is_empty() { + native_children + } else { + satisfied_children.iter().collect() + }; + let candidates: Vec<_> = pool + .into_iter() + .map(|(idx, part, _)| { + let size = + PlanSize::from_plan(children[*idx].context.plan.as_ref()); + (size, *idx, part.clone()) + }) + .collect(); + + // Only select a reference candidate if there is a unique, strictly + // larger winner (`size_a > size_b`). If candidates have equal or + // incomparable sizes (e.g. non-overlapping metrics), return None + // so the optimizer avoids arbitrary tie-breaking and falls back to + // standard distribution. + candidates + .iter() + .find(|(size_a, idx_a, _)| { + size_a.is_known() + && candidates.iter().all(|(size_b, idx_b, _)| { + idx_a == idx_b || size_a > size_b + }) + }) + .map(|(_, idx, part)| (*idx, part.clone())) + } + } + }; + + // Validate that best_satisfied_child can be adapted across all unsatisfied children in the group + let best_satisfied_child = + best_satisfied_child.filter(|(ref_idx, ref_partitioning)| { + unsatisfied_children.iter().all(|&child_idx| { + if child_idx == *ref_idx { + true + } else { + ref_partitioning + .adapt( + &children[child_idx].requirement, + &children[child_idx].context.plan.schema(), + ) + .is_some() + } + }) + }); + let mut changed = false; for child_idx in unsatisfied_children { if repartitioned_for_relationship[child_idx] { continue; } + // If this child is the reference child that we are matching against, do not repartition it. + if let Some((ref_idx, _)) = &best_satisfied_child + && *ref_idx == child_idx + { + continue; + } + let (Distribution::HashPartitioned(exprs) | Distribution::KeyPartitioned(exprs)) = &children[child_idx].requirement else { continue; }; - let already_target_hash = matches!( + // If a satisfied reference child exists, adapt its partitioning to match. + let target_partitioning = + if let Some((ref_idx, ref_partitioning)) = &best_satisfied_child { + if *ref_idx == child_idx { + None + } else { + ref_partitioning.adapt( + &children[child_idx].requirement, + &children[child_idx].context.plan.schema(), + ) + } + } else { + None + }; + + let partitioning = target_partitioning.unwrap_or_else(|| { + Distribution::KeyPartitioned(exprs.to_vec()) + .create_partitioning(target_partitions) + }); + + let already_satisfied = match ( + &partitioning, children[child_idx].context.plan.output_partitioning(), - Partitioning::Hash(_, partition_count) if *partition_count == target_partitions - ) && input_distributions - .child_satisfaction( - child_idx, - children[child_idx].context.plan.as_ref(), - ChildSatisfactionOptions::new(), - )? - .is_satisfied(); - - if already_target_hash { + ) { + (Partitioning::Hash(_, p1), Partitioning::Hash(_, p2)) if p1 == p2 => { + input_distributions + .child_satisfaction( + child_idx, + children[child_idx].context.plan.as_ref(), + ChildSatisfactionOptions::new(), + ) + .map(|s| s.is_satisfied()) + .unwrap_or(false) + } + (Partitioning::Range(r1), Partitioning::Range(r2)) if r1 == r2 => true, + _ => false, + }; + + if already_satisfied { continue; } - let partitioning = Distribution::KeyPartitioned(exprs.to_vec()) - .create_partitioning(target_partitions); - let repartition = RepartitionExec::try_new( - Arc::clone(&children[child_idx].context.plan), - partitioning, - )? - .with_preserve_order(); + let (input, original_child) = if let Some(repart) = children[child_idx] + .context + .plan + .downcast_ref::() + && !children[child_idx].context.children.is_empty() + { + let orig = children[child_idx].context.children.pop().unwrap(); + (Arc::clone(repart.input()), orig) + } else { + let current_plan = Arc::clone(&children[child_idx].context.plan); + let orig = std::mem::replace( + &mut children[child_idx].context, + DistributionContext::new(current_plan, true, vec![]), + ); + (Arc::clone(&orig.plan), orig) + }; + + let repartition = + RepartitionExec::try_new(input, partitioning)?.with_preserve_order(); let plan = Arc::new(repartition) as _; - let original_child = std::mem::replace( - &mut children[child_idx].context, - DistributionContext::new(plan, true, vec![]), - ); - children[child_idx].context.children = vec![original_child]; + children[child_idx].context = + DistributionContext::new(plan, true, vec![original_child]); repartitioned_for_relationship[child_idx] = true; changed = true; } diff --git a/datafusion/physical-plan/src/distribution_requirements.rs b/datafusion/physical-plan/src/distribution_requirements.rs index 6405b1f121ef7..119b2357c02b4 100644 --- a/datafusion/physical-plan/src/distribution_requirements.rs +++ b/datafusion/physical-plan/src/distribution_requirements.rs @@ -136,6 +136,11 @@ impl InputDistributionRequirements { self.children.iter().map(|child| &child.distribution) } + /// Whether these requirements include a co-partitioning relationship between children. + pub fn is_co_partitioned(&self) -> bool { + self.co_partitioned.is_some() + } + /// Return the distribution requirement for a child. pub fn child_distribution(&self, child_idx: usize) -> Option<&Distribution> { self.children diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index e4c6265bce511..ec374b3d62a28 100644 --- a/datafusion/sqllogictest/test_files/range_partitioning.slt +++ b/datafusion/sqllogictest/test_files/range_partitioning.slt @@ -446,8 +446,9 @@ physical_plan ########## # TEST 11: Left-Side Range Hash Joins With Incompatible Range Layouts -# Different split points or partition counts do not satisfy the -# co-partitioned layout requirement. +# Differing split points on equal-size inputs fall back to hash repartitioning, +# while mismatched partition counts adapt the smaller input to match the larger +# input's range partitioning. ########## # Different split points do not satisfy the co-partitioned layout requirement. @@ -478,8 +479,8 @@ ORDER BY l.range_key; 30 300 300 35 350 350 -# Different partition counts do not satisfy the co-partitioned layout -# requirement. +# Different partition counts adapt the smaller input to match the larger +# input's range partitioning. query TT EXPLAIN SELECT l.range_key, l.value, r.value FROM range_partitioned l @@ -487,10 +488,9 @@ LEFT JOIN range_partitioned_narrow r ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] -02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet -04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=3 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=parquet +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--RepartitionExec: partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), input_partitions=3 +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=parquet ########## # TEST 12: LeftMark Subqueries Over Range Hash Joins @@ -907,9 +907,8 @@ reset datafusion.optimizer.subset_repartition_threshold; ########## # TEST 23: Right Join with Mismatched Range Partition Counts Repartitions # Both inputs are range partitioned on range_key, but declare a different number -# of partitions (four vs three). The per-child key requirements can be satisfied -# by Range, but the co-partitioned layout requirement cannot, so Hash -# repartitioning repairs both sides of the right join. +# of partitions (four vs three). The larger input's range partitioning is +# preserved, and the smaller input is repartitioned to match. ########## query TT @@ -919,10 +918,9 @@ RIGHT JOIN range_partitioned_narrow r ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0, range_key@0)], projection=[value@1, range_key@2, value@3] -02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet -04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=3 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=parquet +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--RepartitionExec: partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), input_partitions=3 +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=parquet query III SELECT l.value, r.range_key, r.value