Skip to content

feat(io): Automated Schema Inference and Catalog Exploration for SQL Connectors #257

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
thinkapoorv wants to merge 5 commits into from
Closed

feat(io): Automated Schema Inference and Catalog Exploration for SQL Connectors #257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ed4c727
feat(io): implement automated schema exploration for SQL connectors
thinkapoorv Jul 3, 2026
634ab77
chore(duckdb): allow clippy::struct_excessive_bools on DuckDbWriter
thinkapoorv Jul 3, 2026
3cdabac
Merge branch 'main' of https://github.com/pathwaycom/pathway into fea…
thinkapoorv Jul 13, 2026
19a174c
chore: fix flake8 warnings in test_common.py
thinkapoorv Jul 13, 2026
2652a5d
Refactor automated schema exploration for SQL connectors
thinkapoorv Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
## [Unreleased]

### Added
- `pw.io.postgres.read`, `pw.io.mysql.read`, and `pw.io.mssql.read` now support automated schema exploration. You can omit the `schema` parameter when initializing these database readers, and the engine will dynamically infer the target table's schema (including column types and primary keys) directly from the database catalog at startup. This enables rapid zero-configuration onboarding while maintaining static type validation.
- `pw.io.chroma.write` writes a Pathway table to a [Chroma](https://www.trychroma.com/) collection, keeping the collection in sync with the table as rows are added, changed, and removed. The columns are mapped onto Chroma's record fields explicitly: the optional `primary_key` becomes the record id (when omitted, the row's internal Pathway key is used instead), `embedding` the vector, the optional `document` column the stored text, and `metadata_columns` the record metadata. The collection must already exist. The server is addressed with `host`/`port` (plus `ssl`, `headers`, `tenant`, and `database` for authenticated deployments such as Chroma Cloud).
- `pw.io.qdrant.write` writes a Pathway table to a [Qdrant](https://qdrant.tech/) collection. Each row addition is upserted as a point and each deletion removes the corresponding point, so an update replaces a point rather than duplicating it. The `vector` column supplies the point vector (`list[float]` or a 1-D `numpy.ndarray`) and every other column is stored in the point payload. If the target collection does not exist, it is created on the first write using Cosine distance with the dimension of the written vectors. The `batch_size` parameter bounds how many points are sent per request, and an optional `api_key` authenticates against Qdrant Cloud or a secured instance.
- `pw.io.duckdb.write` writes a Pathway table into a DuckDB database file through a native, in-process connector, in either `"stream_of_changes"` or `"snapshot"` mode. Embeddings stored as `numpy` arrays or `list[float]` columns land in native `DOUBLE[]` list columns, so the result is directly searchable with DuckDB's vector-distance functions for RAG retrieval. The `detach_between_batches` option makes the writer close the database after every minibatch commit and reopen it for the next one, releasing the file lock in between — so a separate process (e.g. a query server) can read committed data with short-lived read-only connections while the pipeline keeps running; the writer retries the reopen with a backoff when a reader momentarily holds the lock.
Expand Down
194 changes: 105 additions & 89 deletions external/differential-dataflow/dogsdogsdogs/examples/delta_query.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,27 +1,25 @@
extern crate timely;
extern crate graph_map;
extern crate differential_dataflow;
extern crate graph_map;
extern crate timely;

extern crate dogsdogsdogs;

use timely::dataflow::Scope;
use timely::dataflow::operators::probe::Handle;
use differential_dataflow::input::Input;
use differential_dataflow::operators::JoinCore;
use graph_map::GraphMMap;
use timely::dataflow::operators::probe::Handle;
use timely::dataflow::Scope;

use dogsdogsdogs::altneu::AltNeu;
use dogsdogsdogs::calculus::{Differentiate, Integrate};

fn main() {

// snag a filename to use for the input graph.
let filename = std::env::args().nth(1).unwrap();
let batching = std::env::args().nth(2).unwrap().parse::<usize>().unwrap();
let inspect = std::env::args().any(|x| x == "inspect");

timely::execute_from_args(std::env::args().skip(2), move |worker| {

let timer = std::time::Instant::now();
let graph = GraphMMap::new(&filename);

Expand All @@ -30,103 +28,121 @@ fn main() {

let mut probe = Handle::new();

let mut input = worker.dataflow::<usize,_,_>(|scope| {

let mut input = worker.dataflow::<usize, _, _>(|scope| {
let (edges_input, edges) = scope.new_collection();

// Graph oriented both ways, indexed by key.
use differential_dataflow::operators::arrange::ArrangeByKey;
let forward_key = edges.arrange_by_key();
let reverse_key = edges.map(|(x,y)| (y,x))
.arrange_by_key();
let reverse_key = edges.map(|(x, y)| (y, x)).arrange_by_key();

// Graph oriented both ways, indexed by (key, val).
use differential_dataflow::operators::arrange::ArrangeBySelf;
let forward_self = edges.arrange_by_self();
let reverse_self = edges.map(|(x,y)| (y,x))
.arrange_by_self();
let reverse_self = edges.map(|(x, y)| (y, x)).arrange_by_self();

// // Graph oriented both ways, counts of distinct vals for each key.
// // Not required without worst-case-optimal join strategy.
// let forward_count = edges.map(|(x,y)| x).arrange_by_self();
// let reverse_count = edges.map(|(x,y)| y).arrange_by_self();

// Q(a,b,c) := E1(a,b), E2(b,c), E3(a,c)
let (triangles_prev, triangles_next) = scope.scoped::<AltNeu<usize>,_,_>("DeltaQuery (Triangles)", |inner| {

// Grab the stream of changes.
let changes = edges.enter(inner);

// Each relation we'll need.
let forward_key_alt = forward_key.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let reverse_key_alt = reverse_key.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let forward_key_neu = forward_key.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
// let reverse_key_neu = reverse_key.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));

// let forward_self_alt = forward_self.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let reverse_self_alt = reverse_self.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let forward_self_neu = forward_self.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
let reverse_self_neu = reverse_self.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));

// For each relation, we form a delta query driven by changes to that relation.
//
// The sequence of joined relations are such that we only introduce relations
// which share some bound attributes with the current stream of deltas.
// Each joined relation is delayed { alt -> neu } if its position in the
// sequence is greater than the delta stream.
// Each joined relation is directed { forward, reverse } by whether the
// bound variable occurs in the first or second position.

let key1 = |x: &(u32, u32)| x.0;
let key2 = |x: &(u32, u32)| x.1;

use dogsdogsdogs::operators::propose;
use dogsdogsdogs::operators::validate;

// Prior technology
// dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
let changes1 = propose(&changes, forward_key_neu.clone(), key2.clone());
let changes1 = validate(&changes1, forward_self_neu.clone(), key1.clone());
let changes1 = changes1.map(|((a,b),c)| (a,b,c));

// dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
let changes2 = propose(&changes, reverse_key_alt.clone(), key1.clone());
let changes2 = validate(&changes2, reverse_self_neu.clone(), key2.clone());
let changes2 = changes2.map(|((b,c),a)| (a,b,c));

// dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
let changes3 = propose(&changes, forward_key_alt.clone(), key1.clone());
let changes3 = validate(&changes3, reverse_self_alt.clone(), key2.clone());
let changes3 = changes3.map(|((a,c),b)| (a,b,c));

let prev_changes = changes1.concat(&changes2).concat(&changes3).leave();

// New ideas
let d_edges = edges.differentiate(inner);

// dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
let changes1 =
d_edges
.map(|(x,y)| (y,x))
.join_core(&forward_key_neu, |b,a,c| Some(((*a, *c), *b)))
.join_core(&forward_self_neu, |(a,c), b, &()| Some((*a,*b,*c)));

// dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
let changes2 =
d_edges
.join_core(&reverse_key_alt, |b,c,a| Some(((*a, *c), *b)))
.join_core(&forward_self_neu, |(a,c), b, &()| Some((*a,*b,*c)));

// dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
let changes3 =
d_edges
.join_core(&forward_key_alt, |a,c,b| Some(((*c, *b), *a)))
.join_core(&reverse_self_alt, |(c,b), a, &()| Some((*a,*b,*c)));

let next_changes = changes1.concat(&changes2).concat(&changes3).integrate();

(prev_changes, next_changes)
});
let (triangles_prev, triangles_next) =
scope.scoped::<AltNeu<usize>, _, _>("DeltaQuery (Triangles)", |inner| {
// Grab the stream of changes.
let changes = edges.enter(inner);

// Each relation we'll need.
let forward_key_alt = forward_key.enter_at(
inner,
|_, _, t| AltNeu::alt(t.clone()),
|t| t.time.saturating_sub(1),
);
let reverse_key_alt = reverse_key.enter_at(
inner,
|_, _, t| AltNeu::alt(t.clone()),
|t| t.time.saturating_sub(1),
);
let forward_key_neu = forward_key.enter_at(
inner,
|_, _, t| AltNeu::neu(t.clone()),
|t| t.time.saturating_sub(1),
);
// let reverse_key_neu = reverse_key.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));

// let forward_self_alt = forward_self.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let reverse_self_alt = reverse_self.enter_at(
inner,
|_, _, t| AltNeu::alt(t.clone()),
|t| t.time.saturating_sub(1),
);
let forward_self_neu = forward_self.enter_at(
inner,
|_, _, t| AltNeu::neu(t.clone()),
|t| t.time.saturating_sub(1),
);
let reverse_self_neu = reverse_self.enter_at(
inner,
|_, _, t| AltNeu::neu(t.clone()),
|t| t.time.saturating_sub(1),
);

// For each relation, we form a delta query driven by changes to that relation.
//
// The sequence of joined relations are such that we only introduce relations
// which share some bound attributes with the current stream of deltas.
// Each joined relation is delayed { alt -> neu } if its position in the
// sequence is greater than the delta stream.
// Each joined relation is directed { forward, reverse } by whether the
// bound variable occurs in the first or second position.

let key1 = |x: &(u32, u32)| x.0;
let key2 = |x: &(u32, u32)| x.1;

use dogsdogsdogs::operators::propose;
use dogsdogsdogs::operators::validate;

// Prior technology
// dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
let changes1 = propose(&changes, forward_key_neu.clone(), key2.clone());
let changes1 = validate(&changes1, forward_self_neu.clone(), key1.clone());
let changes1 = changes1.map(|((a, b), c)| (a, b, c));

// dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
let changes2 = propose(&changes, reverse_key_alt.clone(), key1.clone());
let changes2 = validate(&changes2, reverse_self_neu.clone(), key2.clone());
let changes2 = changes2.map(|((b, c), a)| (a, b, c));

// dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
let changes3 = propose(&changes, forward_key_alt.clone(), key1.clone());
let changes3 = validate(&changes3, reverse_self_alt.clone(), key2.clone());
let changes3 = changes3.map(|((a, c), b)| (a, b, c));

let prev_changes = changes1.concat(&changes2).concat(&changes3).leave();

// New ideas
let d_edges = edges.differentiate(inner);

// dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
let changes1 = d_edges
.map(|(x, y)| (y, x))
.join_core(&forward_key_neu, |b, a, c| Some(((*a, *c), *b)))
.join_core(&forward_self_neu, |(a, c), b, &()| Some((*a, *b, *c)));

// dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
let changes2 = d_edges
.join_core(&reverse_key_alt, |b, c, a| Some(((*a, *c), *b)))
.join_core(&forward_self_neu, |(a, c), b, &()| Some((*a, *b, *c)));

// dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
let changes3 = d_edges
.join_core(&forward_key_alt, |a, c, b| Some(((*c, *b), *a)))
.join_core(&reverse_self_alt, |(c, b), a, &()| Some((*a, *b, *c)));

let next_changes = changes1.concat(&changes2).concat(&changes3).integrate();

(prev_changes, next_changes)
});

// Test if our two methods do the same thing.
triangles_prev.assert_eq(&triangles_next);
Expand Down Expand Up @@ -155,6 +171,6 @@ fn main() {
println!("{:?}\tRound {} complete", timer.elapsed(), index);
}
}

}).unwrap();
})
.unwrap();
}
81 changes: 51 additions & 30 deletions external/differential-dataflow/dogsdogsdogs/examples/delta_query2.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,31 +1,45 @@
extern crate timely;
extern crate graph_map;
extern crate differential_dataflow;
extern crate graph_map;
extern crate timely;

extern crate dogsdogsdogs;

use timely::dataflow::Scope;
use timely::order::Product;
use differential_dataflow::AsCollection;
use timely::dataflow::operators::probe::Handle;
use timely::dataflow::operators::UnorderedInput;
use timely::dataflow::operators::Map;
use differential_dataflow::AsCollection;
use timely::dataflow::operators::UnorderedInput;
use timely::dataflow::Scope;
use timely::order::Product;

fn main() {

timely::execute_from_args(std::env::args().skip(2), move |worker| {

let mut probe = Handle::new();

let (mut i1, mut i2, c1, c2) = worker.dataflow::<usize,_,_>(|scope| {

let (mut i1, mut i2, c1, c2) = worker.dataflow::<usize, _, _>(|scope| {
// Nested scope as `Product<usize, usize>` doesn't refine `()`, because .. coherence.
scope.scoped("InnerScope", |inner| {

use timely::dataflow::operators::unordered_input::UnorderedHandle;

let ((input1, capability1), data1): ((UnorderedHandle<Product<usize, usize>, ((usize, usize), Product<usize, usize>, isize)>, _), _) = inner.new_unordered_input();
let ((input2, capability2), data2): ((UnorderedHandle<Product<usize, usize>, ((usize, usize), Product<usize, usize>, isize)>, _), _) = inner.new_unordered_input();
let ((input1, capability1), data1): (
(
UnorderedHandle<
Product<usize, usize>,
((usize, usize), Product<usize, usize>, isize),
>,
_,
),
_,
) = inner.new_unordered_input();
let ((input2, capability2), data2): (
(
UnorderedHandle<
Product<usize, usize>,
((usize, usize), Product<usize, usize>, isize),
>,
_,
),
_,
) = inner.new_unordered_input();

let edges1 = data1.as_collection();
let edges2 = data2.as_collection();
Expand All @@ -36,50 +50,57 @@ fn main() {
let forward2 = edges2.arrange_by_key();

// Grab the stream of changes. Stash the initial time as payload.
let changes1 = edges1.inner.map(|((k,v),t,r)| ((k,v,t.clone()),t,r)).as_collection();
let changes2 = edges2.inner.map(|((k,v),t,r)| ((k,v,t.clone()),t,r)).as_collection();
let changes1 = edges1
.inner
.map(|((k, v), t, r)| ((k, v, t.clone()), t, r))
.as_collection();
let changes2 = edges2
.inner
.map(|((k, v), t, r)| ((k, v, t.clone()), t, r))
.as_collection();

use dogsdogsdogs::operators::half_join;

// pick a frontier that will not mislead TOTAL ORDER comparisons.
let closure = |time: &Product<usize, usize>| Product::new(time.outer.saturating_sub(1), time.inner.saturating_sub(1));
let closure = |time: &Product<usize, usize>| {
Product::new(time.outer.saturating_sub(1), time.inner.saturating_sub(1))
};

let path1 =
half_join(
let path1 = half_join(
&changes1,
forward2,
closure.clone(),
|t1,t2| t1.lt(t2), // This one ignores concurrent updates.
|t1, t2| t1.lt(t2), // This one ignores concurrent updates.
|key, val1, val2| (key.clone(), (val1.clone(), val2.clone())),
);

let path2 =
half_join(
let path2 = half_join(
&changes2,
forward1,
closure.clone(),
|t1,t2| t1.le(t2), // This one can "see" concurrent updates.
|t1, t2| t1.le(t2), // This one can "see" concurrent updates.
|key, val1, val2| (key.clone(), (val2.clone(), val1.clone())),
);

// Delay updates until the worked payload time.
// This should be at least the ignored update time.
path1.concat(&path2)
.inner.map(|(((k,v),t),_,r)| ((k,v),t,r)).as_collection()
path1
.concat(&path2)
.inner
.map(|(((k, v), t), _, r)| ((k, v), t, r))
.as_collection()
.inspect(|x| println!("{:?}", x))
.probe_with(&mut probe);

(input1, input2, capability1, capability2)
})
});

i1
.session(c1.clone())
i1.session(c1.clone())
.give(((5, 6), Product::new(0, 13), 1));

i2
.session(c2.clone())
i2.session(c2.clone())
.give(((5, 7), Product::new(11, 0), 1));

}).unwrap();
})
.unwrap();
}
Loading
Loading