viirya commented on code in PR #25583:
URL: https://github.com/apache/datafusion/pull/25583#discussion_r4116953812
##########
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:
@alamb Simplified this to describe the observable behavior: adding a metric
does not change other snapshots.
##########
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:
@alamb Agreed. The docs now focus on partition selection, ordering,
duplicates, shared values, and independent membership.
##########
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:
@alamb Updated this to “Metric Registry and Snapshot implementations.”
##########
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:
@alamb Removed these details from the module docs. The necessary invariants
are now documented beside the relevant fields and methods.
##########
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:
@alamb Done. The field is now private, registration goes through a method,
and snapshot creation captures the boundary within this module.
##########
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:
@alamb Removed the extra `Box`. The index is now stored directly in the
registry; its empty HashMap does not allocate buckets until needed.
##########
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:
@alamb Mutation detaches the snapshot before changing its members, leaving
the registry and its index unchanged. I've made this explicit in the branches
and added coverage for mutation after the index has been populated.
##########
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:
@alamb Documented both parameters. `partition` is the partition ID; `end` is
an exclusive position in the full registration vector, including other
partitions and global metrics.
--
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]
