alamb commented on code in PR #25583:
URL: https://github.com/apache/datafusion/pull/25583#discussion_r4086318664
##########
datafusion/physical-expr-common/src/metrics/mod.rs:
##########
@@ -239,11 +250,35 @@ impl MetricsSet {
Default::default()
}
- /// Add the specified metric
+ /// Add the specified metric.
+ ///
+ /// Mutating a deferred snapshot first copies its members into an
independent
Review Comment:
This seems like an implementation detail -- I think the docs on the overall
structure are probably enough
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
+//! catches up with registration on demand; snapshots remember their original
end
+//! position even when a later reader has advanced the index past that
position.
+
+use super::Metric;
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::{Arc, OnceLock};
+
+#[derive(Debug, Default)]
+pub(super) struct Registry {
+ pub(super) metrics: Vec<Arc<Metric>>,
Review Comment:
recommend keeping this private and accessing via a method rather than direct
field access
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
Review Comment:
Could we maybe make this specific tot he code that is in the module --
somethig like
```rust
//! Metric [`Registry`] implementation
```
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
Review Comment:
this is all details of the implementaiton that is probably not useful in the
module level comments
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
+//! catches up with registration on demand; snapshots remember their original
end
+//! position even when a later reader has advanced the index past that
position.
+
+use super::Metric;
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::{Arc, OnceLock};
+
+#[derive(Debug, Default)]
+pub(super) struct Registry {
+ pub(super) metrics: Vec<Arc<Metric>>,
+ // No index allocation or maintenance on the registration path.
+ index: Option<Box<PartitionIndex>>,
Review Comment:
Why is this boxed? The HashMap is already boxed, this seems like it just
adds another allocation 🤔
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
+//! catches up with registration on demand; snapshots remember their original
end
+//! position even when a later reader has advanced the index past that
position.
+
+use super::Metric;
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::{Arc, OnceLock};
+
+#[derive(Debug, Default)]
+pub(super) struct Registry {
+ pub(super) metrics: Vec<Arc<Metric>>,
+ // No index allocation or maintenance on the registration path.
+ index: Option<Box<PartitionIndex>>,
+}
+
+#[derive(Debug, Default)]
+struct PartitionIndex {
+ // Number of registry entries already examined, including global metrics.
+ indexed: usize,
+ // Partition ID -> positions in Registry::metrics, in registration order.
+ positions: HashMap<usize, Vec<usize>>,
+}
+
+impl Registry {
+ pub(super) fn new(metrics: Vec<Arc<Metric>>) -> Self {
+ Self {
+ metrics,
+ index: None,
+ }
+ }
+
+ fn select(&mut self, partition: usize, end: usize) -> Vec<Arc<Metric>> {
+ let index = self.index.get_or_insert_with(Default::default);
+ for position in index.indexed..end {
+ if let Some(partition) = self.metrics[position].partition() {
+ index.positions.entry(partition).or_default().push(position);
+ }
+ }
+ index.indexed = index.indexed.max(end);
+ let Some(positions) = index.positions.get(&partition) else {
+ return Vec::new();
+ };
+ // Another snapshot may already have indexed registrations after our
end.
+ let len = positions.partition_point(|&position| position < end);
+ positions[..len]
+ .iter()
+ .map(|&i| Arc::clone(&self.metrics[i]))
+ .collect()
+ }
+}
+
+#[derive(Clone)]
+pub(super) enum Snapshot {
+ Owned(Vec<Arc<Metric>>),
+ Deferred(Arc<Deferred>),
+}
+
+pub(super) struct Deferred {
+ registry: Arc<Mutex<Registry>>,
+ end: usize,
+ flat: OnceLock<Vec<Arc<Metric>>>,
+}
+
+impl Default for Snapshot {
+ fn default() -> Self {
+ Self::Owned(Vec::new())
+ }
+}
+
+impl fmt::Debug for Snapshot {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_list().entries(self.iter()).finish()
+ }
+}
+
+impl Deferred {
+ fn materialize(&self) -> Vec<Arc<Metric>> {
+ self.registry.lock().metrics[..self.end].to_vec()
+ }
+}
+
+impl Snapshot {
+ pub(super) fn new(registry: Arc<Mutex<Registry>>, end: usize) -> Self {
+ Self::Deferred(Arc::new(Deferred {
+ registry,
+ end,
+ flat: OnceLock::new(),
+ }))
+ }
+
+ pub(super) fn for_partition(&self, partition: usize) -> Self {
+ Self::Owned(match self {
+ Self::Owned(metrics) => metrics
+ .iter()
+ .filter(|m| m.partition() == Some(partition))
Review Comment:
isn't this the filter you were trying to avoid? shouldn't this be using the
index if it is available 😕
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
+//! catches up with registration on demand; snapshots remember their original
end
+//! position even when a later reader has advanced the index past that
position.
+
+use super::Metric;
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::{Arc, OnceLock};
+
+#[derive(Debug, Default)]
+pub(super) struct Registry {
+ pub(super) metrics: Vec<Arc<Metric>>,
+ // No index allocation or maintenance on the registration path.
+ index: Option<Box<PartitionIndex>>,
+}
+
+#[derive(Debug, Default)]
+struct PartitionIndex {
+ // Number of registry entries already examined, including global metrics.
+ indexed: usize,
+ // Partition ID -> positions in Registry::metrics, in registration order.
+ positions: HashMap<usize, Vec<usize>>,
Review Comment:
Since the Arc's are only a few more pointers, I wonder if you considered
having this `HashMap<usize, Arc<Metric>` or something, so getting the
partition's metrics would be an update to the index, and then a clone of the
relevant Vec
##########
datafusion/physical-expr-common/src/metrics/mod.rs:
##########
@@ -239,11 +250,35 @@ impl MetricsSet {
Default::default()
}
- /// Add the specified metric
+ /// Add the specified metric.
+ ///
+ /// Mutating a deferred snapshot first copies its members into an
independent
+ /// vector. Subsequent additions append to that vector.
pub fn push(&mut self, metric: Arc<Metric>) {
self.metrics.push(metric)
}
+ /// Return a snapshot containing only metrics with `partition ==
Some(partition)`.
+ ///
+ /// For registry-backed snapshots, this incrementally indexes registrations
Review Comment:
this likewise seems like a bunch of implementation specific detail -- I
think it would help if the comments only focused on the end user visible
behavior. I am not sure how to interpret all the stuff about cloning matching
handles, et
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
+//! catches up with registration on demand; snapshots remember their original
end
+//! position even when a later reader has advanced the index past that
position.
+
+use super::Metric;
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::{Arc, OnceLock};
+
+#[derive(Debug, Default)]
+pub(super) struct Registry {
+ pub(super) metrics: Vec<Arc<Metric>>,
+ // No index allocation or maintenance on the registration path.
+ index: Option<Box<PartitionIndex>>,
+}
+
+#[derive(Debug, Default)]
+struct PartitionIndex {
+ // Number of registry entries already examined, including global metrics.
+ indexed: usize,
+ // Partition ID -> positions in Registry::metrics, in registration order.
+ positions: HashMap<usize, Vec<usize>>,
+}
+
+impl Registry {
+ pub(super) fn new(metrics: Vec<Arc<Metric>>) -> Self {
+ Self {
+ metrics,
+ index: None,
+ }
+ }
+
+ fn select(&mut self, partition: usize, end: usize) -> Vec<Arc<Metric>> {
Review Comment:
could we please document what partition and end mean in this? Is the end
relative to just the metrics in the partition? or all the metrics?
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
+//! catches up with registration on demand; snapshots remember their original
end
+//! position even when a later reader has advanced the index past that
position.
+
+use super::Metric;
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::{Arc, OnceLock};
+
+#[derive(Debug, Default)]
+pub(super) struct Registry {
+ pub(super) metrics: Vec<Arc<Metric>>,
+ // No index allocation or maintenance on the registration path.
+ index: Option<Box<PartitionIndex>>,
+}
+
+#[derive(Debug, Default)]
+struct PartitionIndex {
+ // Number of registry entries already examined, including global metrics.
+ indexed: usize,
+ // Partition ID -> positions in Registry::metrics, in registration order.
+ positions: HashMap<usize, Vec<usize>>,
+}
+
+impl Registry {
+ pub(super) fn new(metrics: Vec<Arc<Metric>>) -> Self {
+ Self {
+ metrics,
+ index: None,
+ }
+ }
+
+ fn select(&mut self, partition: usize, end: usize) -> Vec<Arc<Metric>> {
+ let index = self.index.get_or_insert_with(Default::default);
+ for position in index.indexed..end {
+ if let Some(partition) = self.metrics[position].partition() {
+ index.positions.entry(partition).or_default().push(position);
+ }
+ }
+ index.indexed = index.indexed.max(end);
+ let Some(positions) = index.positions.get(&partition) else {
+ return Vec::new();
+ };
+ // Another snapshot may already have indexed registrations after our
end.
+ let len = positions.partition_point(|&position| position < end);
+ positions[..len]
+ .iter()
+ .map(|&i| Arc::clone(&self.metrics[i]))
+ .collect()
+ }
+}
+
+#[derive(Clone)]
+pub(super) enum Snapshot {
+ Owned(Vec<Arc<Metric>>),
+ Deferred(Arc<Deferred>),
+}
+
+pub(super) struct Deferred {
+ registry: Arc<Mutex<Registry>>,
+ end: usize,
+ flat: OnceLock<Vec<Arc<Metric>>>,
+}
+
+impl Default for Snapshot {
+ fn default() -> Self {
+ Self::Owned(Vec::new())
+ }
+}
+
+impl fmt::Debug for Snapshot {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_list().entries(self.iter()).finish()
+ }
+}
+
+impl Deferred {
+ fn materialize(&self) -> Vec<Arc<Metric>> {
+ self.registry.lock().metrics[..self.end].to_vec()
+ }
+}
+
+impl Snapshot {
+ pub(super) fn new(registry: Arc<Mutex<Registry>>, end: usize) -> Self {
+ Self::Deferred(Arc::new(Deferred {
+ registry,
+ end,
+ flat: OnceLock::new(),
+ }))
+ }
+
+ pub(super) fn for_partition(&self, partition: usize) -> Self {
+ Self::Owned(match self {
+ Self::Owned(metrics) => metrics
+ .iter()
+ .filter(|m| m.partition() == Some(partition))
+ .cloned()
+ .collect(),
+ Self::Deferred(snapshot) => {
+ snapshot.registry.lock().select(partition, snapshot.end)
+ }
+ })
+ }
+
+ pub(super) fn push(&mut self, metric: Arc<Metric>) {
+ if let Self::Deferred(_) = self {
+ *self = Self::Owned(std::mem::take(self).into_iter().collect());
+ }
+ let Self::Owned(metrics) = self else {
+ unreachable!()
+ };
+ metrics.push(metric);
Review Comment:
doesn't this also invalidate the partition index, if there is one?
##########
datafusion/physical-expr-common/src/metrics/snapshot.rs:
##########
@@ -0,0 +1,301 @@
+// 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.
+
+//! Fixed-membership snapshots of an append-only registry.
+//!
+//! Registration only appends to a vector. Partition readers share an index
that
+//! catches up with registration on demand; snapshots remember their original
end
+//! position even when a later reader has advanced the index past that
position.
+
+use super::Metric;
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::{Arc, OnceLock};
+
+#[derive(Debug, Default)]
+pub(super) struct Registry {
Review Comment:
I am sorry I don't understand this design -- maybe we can comment why bother
creating the `PartitionIndex` at all? It seems like it just adds overhead (a
new hash map and a bunch of allocations)
--
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]
