Re: [PR] Add range partitioning enum variant [datafusion]
alamb commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3274516944
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +137,176 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes an ordered key space with split points.
+///
+/// - `sort_exprs` define the partitioning key and ordering.
Review Comment:
great
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +137,176 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes an ordered key space with split points.
+///
+/// - `sort_exprs` define the partitioning key and ordering.
+/// - `split_points` define the boundaries between adjacent partitions. Each
+/// split point is a tuple with one [`ScalarValue`] per sort expression.
+/// - The declaring source must ensure every emitted row belongs to exactly one
+/// declared partition and is emitted by that partition.
+///
+/// The sort expressions must be non-empty, and split points must be strictly
+/// ordered according to those sort expressions.
+///
+/// For a single range key:
+///
+/// ```text
+/// sort_exprs = [date ASC NULLS LAST]
+/// split_points = [
+/// (2022-01-01),
+/// (2023-01-01),
+/// ]
+///
+/// partition 0: date before 2022-01-01
+/// partition 1: date between 2022-01-01 and 2023-01-01
Review Comment:
I think we should carefully and formally define the edge conditions and how
split poins define partitons. I think the idea is that idea is that each value
is `<=` its split point
SO for example, I think that would mean that `N` split points defines `N+1`
partitions and the key range would be divided up as follows across partitions
* Partiton 0: `key < split_points[0]`
* Partiton 1: `split_points[0] <= key < split_points[1]`
* ...
* Partition N-1: `split_points[N-2] <= key < split_points[N-1]`
* Partition N: `split_points[N-1] < key`
```suggestion
/// partition 1: date between 2022-01-01 (inclusive) and 2023-01-01
(exclusive)
```
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +137,176 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes an ordered key space with split points.
+///
+/// - `sort_exprs` define the partitioning key and ordering.
+/// - `split_points` define the boundaries between adjacent partitions. Each
+/// split point is a tuple with one [`ScalarValue`] per sort expression.
+/// - The declaring source must ensure every emitted row belongs to exactly one
+/// declared partition and is emitted by that partition.
Review Comment:
I think the new definition defines a clear partition for every row -- so it
is not possble to emit a row that does not belong to a partition
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -1482,6 +1492,11 @@ impl ExecutionPlan for RepartitionExec {
if !self.maintains_input_order()[0] {
return Ok(SortOrderPushdownResult::Unsupported);
}
+if matches!(self.partitioning(), Partitioning::Range(_)) {
Review Comment:
I recommend making using an explict match here so it is clear what is
supported nad what is not (and the guard is less likely to be removed
accidentally)
```rust
match self.partitionoing {
...
}
```
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +137,176 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+///
Re: [PR] Add range partitioning enum variant [datafusion]
alamb commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3274513560
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
Review Comment:
It seems I was late to the party and @gene-bordegaray just did this in
e4c860f
Checking out the new one
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
alamb commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3274121399
##
datafusion/ffi/src/physical_expr/partitioning.rs:
##
@@ -45,6 +45,10 @@ impl From<&Partitioning> for FFI_Partitioning {
.collect();
Self::Hash(exprs, *size)
}
+// FFI does not yet expose range partition metadata.
Review Comment:
When we merge this PR, it might help to make github issues for each of these
subtask / TODOs and add a link in the code. That way if people stumble across
the gap in the code they can find the issue and we would have a nice list of
tasks to do (either for other contributors of for coding agents)
SOmething like
```rust
// FFI does not yet expose range partition metadata.
// See https://github.com/apache/datafusion/issues/XYZ
```
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
> I foudn that they store physical partitioning metadata, but did not find
anything like a “multi-dimensional repartition this row.”
I agree -- Influx's model is best modeled as "compound key" (it is not
multi-dimensional partitioning)
> So I think compound-key range partitioning is the right move. If there is
a use for this I would say that this should be its own separate implementation.
I agree
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
+///
+/// NOTE: Optimizer and execution behavior for this partitioning is
intentionally
+/// not implemented and will be introduced incrementally. This public API keeps
+/// the partition ranges explicit for users. Repartitioning may compile the
same
+/// metadata into a more efficient internal router.
+#[derive(Debug, Clone)]
+pub struct RangePartitioning {
+partition_exprs: Vec>,
+partitions: Vec,
+}
+
+impl RangePartitioning {
+/// Creates range partitioning metadata.
+///
+/// The caller is responsible for ensuring each partition has one
Re: [PR] Add range partitioning enum variant [datafusion]
alamb commented on PR #22207: URL: https://github.com/apache/datafusion/pull/22207#issuecomment-4498653192 Note there is more discussion here - https://github.com/apache/datafusion/issues/21992#issuecomment-4466649802 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
gene-bordegaray commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3273970550
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
I looked at ClickHouse and InfluxDB, I foudn that they store physical
partitioning metadata, but did not find anything like a “multi-dimensional
repartition this row.”
I looked into systems that try to do a true multi-dimensional partitioning
and there aren't many that really do it. I think fo good reason. It would treat
the columns like `time` and `city` as independent axes, which in simple cases
is great and easy but when things start to overlap or more nuanced it seem we
would need a routing structure like a grid/sparse map/KDB-tree (these were very
complicated).
The closest thing I found was in Sedona where they do spatial partitioning
using quadtree and kdbtree:
-
https://sedona.apache.org/1.7.1/api/rdocs/reference/sedona_apply_spatial_partitioner.html
- Quadtree: https://www.geeksforgeeks.org/dsa/quad-tree/
- KDBTree: https://en.wikipedia.org/wiki/K-D-B-tree
With compound-key range partitioning it is more clear and still efficient on
repartition routing:
```text
1. evaluate `(time, city)` as one ordered key
2. binary-search split points
3. route to a partition.
```
Compound-key range partitioning should cover most join/planner cases like
@stuhood mentioned. We are typicaly asking "are the two sides of this join
compatible" for things like dynamic filters. The thing it lacks compare to true
multi-dimensional partitioning is independent routing. So, for example, it
cannot directly represent “time bucket X and city bucket Y map to partition P”
which is useful when we want to do optimizations on each axis independently
like pruning on the individual columns:
```sql
WHERE time >= '2022' AND time < '2023'
AND city >= 'Boston' AND city < 'NYC'
```
So I think compound-key range partitioning is the right move. If there is a
use for this I would say that this should be its own separate implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
NGA-TRAN commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3262085232
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
Yeah, maybe look to see how other DBs support this. I know ClickHouse and
InfluxDB do
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
stuhood commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3261654831
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
Ok, interesting. Yea, if there are multiple consumers who are interested in
multi-dimensional partitioning, and it can still reduce down to a base-case of
single-dimension partitioning for consumers who don't need that complexity,
then perhaps it could make sense to bake it in here.
I'll be honest though: my largest concern is just that I have no experience
with multi: only single. So I have less useful feedback to give.
One thing that could likely be a good exercise in terms of the
representation would be figuring out what datastructure you would/could use to
efficiently partition in multiple dimensions, and then bias towards a
representation which allows you to construct that datastructure. In one
dimensional partitioning, that's essentially just a binary-tree/b-tree/sorted
structure: hence the desire for non-overlapping contiguous ranges (to avoid
needing something more complex like an interval tree). For multi-dimensional
partitioning, what structure would you use, and what would the inputs to
construct one be? I expect that fully covering the space (contiguous,
no-overlap) makes the multi-dimensional datastructure cheaper/simpler as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
stuhood commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3261654831
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
Ok, interesting. Yea, if there are multiple consumers who are interested in
multi-dimensional partitioning, and it can still reduce down to a base-case of
single-dimension partitioning for consumers who don't need that complexity,
then perhaps it could make sense to bake it in here.
I'll be honest though: my largest concern is just that I have no experience
with multi: only single. So I have less useful feedback to give.
One thing that could likely be a good exercise in terms of the
representation would be figuring out what datastructure you would/could use to
efficiently partition in multiple dimensions, and then bias towards a
representation which allows you to construct that datastructure. In one
dimensional partitioning, that's essentially just a binary-tree/b-tree/sorted
structure: hence the desire for non-overlapping contiguous ranges (to avoid
needing something more complex like an interval tree). For multi-dimensional
partitioning, what structure would you use, and what would the inputs to
construct one be? I expect that fully covering the space makes that cheaper as
well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
stuhood commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3261654831
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
Ok, interesting. Yea, if there are multiple consumers who are interested in
multi-dimensional partitioning, and it can still reduce down to a base-case of
single-dimension partitioning for consumers who don't need that complexity,
then perhaps it could make sense to bake it in here.
I'll be honest though: my largest concern is just that I have no experience
with multi: only single. So I have less useful feedback to give.
One thing that could likely be a good exercise in terms of the
representation would be figuring out what datastructure you would/could use to
efficiently partition in multiple dimensions, and then bias towards a
representation which allows you to construct that datastructure. In one
dimensional partitioning, that's essentially just a binary-tree/b-tree/sorted
structure: hence the desire for non-overlapping contiguous ranges (to avoid
needing something more complex like an interval tree). For multi-dimensional
partitioning, what structure would you use, and what would the inputs to
construct one be? I expect that fully covering the space (contiguous,
no-overlap) makes that tree cheaper as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
gene-bordegaray commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3261586069
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
+///
+/// NOTE: Optimizer and execution behavior for this partitioning is
intentionally
+/// not implemented and will be introduced incrementally. This public API keeps
+/// the partition ranges explicit for users. Repartitioning may compile the
same
+/// metadata into a more efficient internal router.
+#[derive(Debug, Clone)]
+pub struct RangePartitioning {
+partition_exprs: Vec>,
+partitions: Vec,
+}
+
+impl RangePartitioning {
+/// Creates range partitioning metadata.
+///
+/// The caller is responsible for ensuring each partition has one range per
+/// partition expression and for satisfying the contract documented on
+/// [`RangePartitioning`].
+pub fn new(
+partition_exprs: Vec>,
+partitions: Vec,
+) -> Self {
+Self {
+partition_exprs,
+partitions,
+}
+}
+
+/// Returns the partition expressions.
+pub fn partition_exprs(&self) -> &[Arc] {
+&self.partition_exprs
+}
+
+/// Returns the declared range partitions.
+pub fn partitions(&self) -> &[RangePartition] {
+&self.partitions
+}
+
+/// Returns the number of partitions.
+pub fn partition_count(&self) -> usize {
+self.partitions.len()
+}
+
+fn project(
+&self,
+mapping: &ProjectionMapping,
+input_eq_properties: &EquivalenceProperties,
+) -> Option {
+let partition_exprs = input_eq_properties
+.project_expressions(&self.partition_exprs, mapping)
+.collect::>>()?;
+
+Some(Self {
+partition_exprs,
+partitions: self.partitions.clone(),
+})
+}
+}
+
+impl Display for RangePartitioning {
+fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+let partitions = self
+.partitions
+.iter()
+.map(|partition| format!("{partition}"))
+.collect::>()
+.join(", ");
+write!(
+f,
+"Range({}, [{}], {})",
+format_physical_expr_list(&self.partition_exprs),
+partitions,
+self.partition_count()
+)
+}
+}
+
+impl PartialEq for RangePartitioning {
+fn eq(&self, other: &Self) -> bool {
+physical_exprs_equal(&self.partition_exprs, &other.partition_exprs)
+&& self.partitions == other.partitions
+}
+}
+
+/// Ranges for one output partition in a [`RangePartitioning`].
+#[derive(Debug, Clone, PartialEq)]
+pub struct RangePartition {
+ranges: Vec,
+}
+
+impl RangePartition {
+/// Creates a partition from one range per partition expression.
+pub fn new(ranges: Vec) -> Self {
+Self { ranges }
+}
+
+/// Returns the ranges for this partition.
+pub fn ranges(&self) -> &[RangeInterval] {
+&self.ranges
+}
+}
+
+impl Display for RangePartition {
+fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+let ranges = self
+.ranges
+.iter()
+.map(|range| format!("{range}"))
+.collect::>()
+.join(", ");
+write!(f, "({ranges})")
+}
+}
+
+/// A scalar interval in one range partition dimension.
+#[derive(Debug, Clone, PartialEq)]
+pub struct RangeInterval {
+lower: Option,
+upper: Option,
Review Comment:
Re: [PR] Add range partitioning enum variant [datafusion]
gene-bordegaray commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3261514072
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
This is supposed to be multi-dimensional. For example we are partitoned on
independent `id`, `time` keys thus this would accurately represent our layout.
I do see what you are saying about the join needing a single key, which will
work for our case as well. But maybe this can start as single dimension with
compound keys and extend if the use case arises to avoid complexity?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add range partitioning enum variant [datafusion]
stuhood commented on code in PR #22207:
URL: https://github.com/apache/datafusion/pull/22207#discussion_r3261280186
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Allston, Boston)
+///
+/// partition 1:
+/// date in [2021-01-01, 2022-01-01)
+/// city in [Boston, NYC)
+/// ```
Review Comment:
Are these supposed to be representing compound keys or multi-dimensional
partitioning?
If they are compound keys, then I think that it would be clearer to express
them as:
```
[
(2021-01-01, Allston),
(2022-01-01, Boston),
...
]
```
If this is supposed to be multi-dimensional partitioning, then I think that
that might be unnecessary, as mentioned on the discussion thread: any
particular join only needs to consider 1 dimension (possibly with compound
keys).
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
Review Comment:
It would probably be good to figure out the null semantics early... which I
think could involve baking a `SortOptions` struct in here (or at least
nulls-first/nulls-last).
##
datafusion/physical-expr/src/partitioning.rs:
##
@@ -133,13 +136,225 @@ impl Display for Partitioning {
.join(", ");
write!(f, "Hash([{phy_exprs_str}], {size})")
}
+Partitioning::Range(range) => write!(f, "{range}"),
Partitioning::UnknownPartitioning(size) => {
write!(f, "UnknownPartitioning({size})")
}
}
}
}
+/// Physical range partitioning.
+///
+/// [`RangePartitioning`] describes range bounds over one or more physical
+/// expressions. Each [`RangePartition`] represents one output partition and
must
+/// contain exactly one [`RangeInterval`] for each partition expression.
+///
+/// The source declaring this partitioning is responsible for ensuring that,
for
+/// every emitted row, the row belongs to exactly one partition and is emitted
by
+/// that partition. The declared ranges do not need to cover values that the
plan
+/// cannot emit.
+///
+/// Each lower and upper bound explicitly records whether it is inclusive.
+/// Unbounded sides are represented with `None`, bound values should be
non-null
+/// until null routing semantics are defined.
+///
+/// For example, a scan can declare date and city range partitions as:
+///
+/// ```text
+/// exprs = [date, city]
+///
+/// partition 0:
+/// date in [2021-01-01, 2022-01-01)
+/// city in
