Skip to content

SS-456 - storage: give each PostgreSQL source export its own output port #39418

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

Merged
martykulma merged 2 commits into from
Oct 14, 2026
Merged

SS-456 - storage: give each PostgreSQL source export its own output port #39418

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
adb37ec
storage: give each PostgreSQL source export its own output port
martykulma Sep 30, 2026
28812a2
PR feedback
martykulma Oct 7, 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
38 changes: 15 additions & 23 deletions src/storage/src/source/postgres.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,23 +76,25 @@
//! │concat│ │
//! ╰──┬───╯ │
//! │ data │progress
//! │ output │output
//! │ outputs │output
//! │ (one per export) │
//! v v
//! ```
//!
//! Both the snapshot and the replication operators emit one output stream per source export, and
//! the streams of each export are concatenated pairwise.

use std::collections::BTreeMap;
use std::rc::Rc;
use std::time::Duration;

use differential_dataflow::AsCollection;
use itertools::Itertools as _;
use mz_expr::EvalError;
use mz_ore::cast::CastFrom;
use mz_ore::error::ErrorExt;
use mz_postgres_util::desc::PostgresTableDesc;
use mz_postgres_util::schema_change::SchemaChangeError;
use mz_postgres_util::{Client, PostgresError, Sql, query_opt, simple_query_opt, sql};
use mz_repr::{Datum, Diff, GlobalId, Row};
use mz_repr::{Datum, GlobalId, Row};
use mz_storage_types::errors::{DataflowError, SourceError, SourceErrorDetails};
use mz_storage_types::sources::casts::StorageScalarExpr;
use mz_storage_types::sources::postgres::CastType;
Expand All @@ -101,9 +103,7 @@ use mz_storage_types::sources::{
};
use mz_timely_util::builder_async::PressOnDropButton;
use serde::{Deserialize, Serialize};
use timely::container::CapacityContainerBuilder;
use timely::dataflow::operators::Concat;
use timely::dataflow::operators::core::Partition;
use timely::dataflow::operators::vec::{Map, ToStream};
use timely::dataflow::{Scope, StreamVec};
use timely::progress::{Antichain, Timestamp};
Expand Down Expand Up @@ -199,24 +199,16 @@ impl SourceRender for PostgresSourceConnection {
metrics,
);

let updates = snapshot_updates.concat(repl_updates);
let partition_count = u64::cast_from(config.source_exports.len());
let data_streams: Vec<_> = updates
.inner
.partition::<CapacityContainerBuilder<_>, _, _>(
partition_count,
|((output, data), time, diff): (
(usize, Result<SourceMessage, DataflowError>),
MzOffset,
Diff,
)| {
let output = u64::cast_from(output);
(output, (data, time, diff))
},
);
// Both operators produce one stream per export, in output index order, which matches the
// iteration order of `source_exports`.
let mut data_collections = BTreeMap::new();
for (id, data_stream) in config.source_exports.keys().zip_eq(data_streams) {
data_collections.insert(*id, data_stream.as_collection());
for ((id, snapshot_updates), repl_updates) in config
.source_exports
.keys()
.zip_eq(snapshot_updates)
.zip_eq(repl_updates)
{
data_collections.insert(*id, snapshot_updates.concat(repl_updates));
}

let export_ids = config.source_exports.keys().copied();
Expand Down
Loading
Loading