Skip to content

Commit 9bf97ad

Browse files
committed
Merge ji/exec-3-scan-concat-pack-nodes: no mid-run compilation
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
2 parents 78b728f + 859a430 commit 9bf97ad

2 files changed

Lines changed: 0 additions & 45 deletions

File tree

‎vortex-layout/src/plan/pipeline/mod.rs‎

Lines changed: 0 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,6 @@ use vortex_array::ArrayRef;
4646
use vortex_array::ExecutionCtx;
4747
use vortex_array::buffer::BufferHandle;
4848
use vortex_error::VortexResult;
49-
use vortex_mask::Mask;
5049
use vortex_session::VortexSession;
5150

5251
pub use self::compile::Chain;
@@ -61,7 +60,6 @@ pub use self::scan::ReadRequest;
6160
pub use self::scan::Scan;
6261
pub use self::scan::Split;
6362
pub use self::scan::Turn;
64-
use crate::plan::PlanRef;
6563
use crate::segments::SegmentId;
6664

6765
/// Batches an inlet holds before its writer is blocked, unless its reader says otherwise.
@@ -140,13 +138,6 @@ pub trait Source: Operator {
140138
}
141139
}
142140

143-
/// A plan a source asks to have compiled into a new inlet.
144-
pub(crate) struct Spawn {
145-
plan: PlanRef,
146-
rows: std::ops::Range<u64>,
147-
mask: Mask,
148-
}
149-
150141
/// What a stage may touch during one call: its pipeline's inlets, the bytes it asked for, and
151142
/// the session.
152143
pub struct Cx<'a> {
@@ -155,7 +146,6 @@ pub struct Cx<'a> {
155146
bytes: &'a mut Option<BufferHandle>,
156147
session: &'a VortexSession,
157148
exec: &'a mut ExecutionCtx,
158-
spawns: &'a mut Vec<Spawn>,
159149
/// The inlets read during this run, so only their writers are checked for room after it.
160150
touched: &'a mut SmallVec<[usize; 4]>,
161151
}
@@ -183,16 +173,6 @@ impl Cx<'_> {
183173
pub fn exec(&mut self) -> &mut ExecutionCtx {
184174
self.exec
185175
}
186-
187-
/// Asks for `plan` over `rows`, restricted to `mask`, to be compiled into a new inlet of this
188-
/// source once the call returns, and returns the inlet's index.
189-
///
190-
/// For a source that learns what it must read only from what it has read, as a list learns
191-
/// its elements' range from its offsets.
192-
pub fn spawn(&mut self, plan: PlanRef, rows: std::ops::Range<u64>, mask: Mask) -> usize {
193-
self.spawns.push(Spawn { plan, rows, mask });
194-
self.inlets.len() + self.spawns.len() - 1
195-
}
196176
}
197177

198178
#[cfg(test)]

‎vortex-layout/src/plan/pipeline/scan.rs‎

Lines changed: 0 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ use super::Cx;
2626
use super::Input;
2727
use super::Operator;
2828
use super::Source;
29-
use super::Spawn;
3029
use super::Step;
3130
use super::compile::Chain;
3231
use super::compile::Shares;
@@ -97,7 +96,6 @@ pub(crate) struct Pipeline {
9796
inlets: Vec<PortId>,
9897
outlets: SmallVec<[PortId; 1]>,
9998
bytes: Option<BufferHandle>,
100-
spawns: Vec<Spawn>,
10199
/// The inlets the last run read.
102100
touched: SmallVec<[usize; 4]>,
103101
state: State,
@@ -129,7 +127,6 @@ fn drive(
129127
inlets,
130128
outlets,
131129
bytes,
132-
spawns,
133130
touched,
134131
..
135132
} = pipe;
@@ -139,7 +136,6 @@ fn drive(
139136
bytes,
140137
session,
141138
exec,
142-
spawns,
143139
touched,
144140
};
145141
loop {
@@ -283,7 +279,6 @@ impl Core {
283279
inlets: chain.inlets,
284280
outlets,
285281
bytes: None,
286-
spawns: Vec::new(),
287282
touched: SmallVec::new(),
288283
state: State::Runnable,
289284
queued: true,
@@ -355,11 +350,6 @@ impl Core {
355350
.ok_or_else(|| vortex_err!("Pipeline {id} is not live"))?;
356351
pipe.queued = false;
357352
let progress = drive(pipe, &mut self.arena, &self.session, &mut self.exec)?;
358-
if !pipe.spawns.is_empty() {
359-
for spawn in std::mem::take(&mut pipe.spawns) {
360-
self.spawn_inlet(id, spawn)?;
361-
}
362-
}
363353
match progress {
364354
Progress::Read(segment_id) => {
365355
let read = ReadId(self.next_read);
@@ -384,21 +374,6 @@ impl Core {
384374
Ok(())
385375
}
386376

387-
/// Compiles what a source asked for into a new inlet of it.
388-
fn spawn_inlet(&mut self, id: PipelineId, spawn: Spawn) -> VortexResult<()> {
389-
let index = self.pipeline(id).inlets.len();
390-
let capacity = self.pipeline(id).source.capacity(index);
391-
let port = self.arena.create(capacity, None, Reader::Pipeline(id));
392-
self.pipeline_mut(id).inlets.push(port);
393-
match self.compile(&spawn.plan, spawn.rows, &spawn.mask)? {
394-
Some(chain) => {
395-
self.add_pipeline(chain, smallvec::smallvec![port]);
396-
}
397-
None => self.arena.close(port),
398-
}
399-
Ok(())
400-
}
401-
402377
/// Wakes the readers of what pipeline `id` wrote and the writers of what it took.
403378
fn wake_neighbours(&mut self, id: PipelineId) {
404379
let mut wake: SmallVec<[PipelineId; 4]> = SmallVec::new();

0 commit comments

Comments
 (0)