Skip to content

Commit

Permalink
feat: support alter shared source
Browse files Browse the repository at this point in the history
- table with connector: filled when creating job catalog https://github.com/risingwavelabs/risingwave/blob/193e93fd8d9f9dbae717fe6a5b411e7f33382f27/src/meta/src/controller/streaming_job.rs#L247-L251
- stream node: filled in fill_job

Signed-off-by: xxchan <[email protected]>

feat: support alter shared source

.

Signed-off-by: xxchan <[email protected]>
  • Loading branch information
xxchan committed Nov 22, 2024
1 parent cd83140 commit 87e92dd
Show file tree
Hide file tree
Showing 12 changed files with 230 additions and 22 deletions.
138 changes: 138 additions & 0 deletions e2e_test/source_inline/kafka/alter/add_column_shared.slt
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
control substitution on

system ok
rpk topic create shared_source_alter -p 4

system ok
cat << EOF | rpk topic produce shared_source_alter -f "%p %v\n" -p 0
0 {"v1": 1, "v2": "a", "v3": "a1"}
1 {"v1": 2, "v2": "b", "v3": "b1"}
2 {"v1": 3, "v2": "c", "v3": "c1"}
3 {"v1": 4, "v2": "d", "v3": "d1"}
EOF

statement ok
create source s (v1 int, v2 varchar, v3 varchar) with (
${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
topic = 'shared_source_alter',
scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;


statement ok
create materialized view mv_before_alter as select * from s;

sleep 2s

query ?? rowsort
select * from s;
----
1 a
2 b
3 c
4 d

query ?? rowsort
select * from mv_before_alter;
----
1 a
2 b
3 c
4 d


statement ok
alter source s add column v3 varchar;

# New MV will have v3.

statement ok
create materialized view mv_after_alter as select * from s;

query ??? rowsort
select * from mv_after_alter;
----
1 a a1
2 b b1
3 c c1
4 d d1

# Batch select from source will have v3.

query ??? rowsort
select * from s;
----
1 a a1
2 b b1
3 c c1
4 d d1

# Old MV is not affected.

query ?? rowsort
select * from mv_before_alter;
----
1 a
2 b
3 c
4 d

# Produce new data.

system ok
cat << EOF | rpk topic produce shared_source_alter -f "%p %v\n" -p 0
0 {"v1": 5, "v2": "e", "v3": "e1"}
1 {"v1": 6, "v2": "f", "v3": "f1"}
2 {"v1": 7, "v2": "g", "v3": "g1"}
3 {"v1": 8, "v2": "h", "v3": "h1"}
EOF

sleep 2s


query ??? rowsort
select * from mv_after_alter;
----
1 a a1
2 b b1
3 c c1
4 d d1
5 e e1
6 f f1
7 g g1
8 h h1


# Batch select from source will have v3.

query ??? rowsort
select * from s;
----
1 a a1
2 b b1
3 c c1
4 d d1
5 e e1
6 f f1
7 g g1
8 h h1

# Old MV is not affected.

query ?? rowsort
select * from mv_before_alter;
----
1 a
2 b
3 c
4 d
5 e
6 f
7 g
8 h


statement ok
drop source s cascade;

# TODO: test alter source with schema registry
14 changes: 14 additions & 0 deletions proto/ddl_service.proto
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,8 @@ message DropSourceResponse {

message AlterSourceRequest {
catalog.Source source = 1;
// for shared source, we need to replace the streaming job
optional ReplaceStreamingJobPlan plan = 2;
}

message AlterSourceResponse {
Expand Down Expand Up @@ -368,6 +370,18 @@ message ReplaceTablePlan {
TableJobType job_type = 5;
}

// Replace a streaming job, but not a table. e.g., alter a shared source.
message ReplaceStreamingJobPlan {
// The new materialization plan, where all schema are updated.
stream_plan.StreamFragmentGraph fragment_graph = 2;
// The mapping from the old columns to the new columns of the table.
// If no column modifications occur (such as for sinking into table), this will be None.
catalog.ColIndexMapping table_col_index_mapping = 3;
// Source catalog of table's associated source
catalog.Source source = 4;
TableJobType job_type = 5;
}

message ReplaceTablePlanRequest {
ReplaceTablePlan plan = 1;
}
Expand Down
5 changes: 3 additions & 2 deletions proto/stream_plan.proto
Original file line number Diff line number Diff line change
Expand Up @@ -116,19 +116,20 @@ message BarrierMutation {
// Stop a set of actors, used for dropping materialized views. Empty dispatchers will be
// automatically removed.
StopMutation stop = 4;
// Update outputs and hash mappings for some dispatchers, used for scaling.
// Update outputs and hash mappings for some dispatchers, used for scaling and replace table.
UpdateMutation update = 5;
// Change the split of some sources.
SourceChangeSplitMutation splits = 6;
// Pause the dataflow of the whole streaming graph, only used for scaling.
PauseMutation pause = 7;
// Resume the dataflow of the whole streaming graph, only used for scaling.
ResumeMutation resume = 8;
// Throttle specific source exec or chain exec.
// Throttle specific source exec or backfill exec.
ThrottleMutation throttle = 10;
// Drop subscription on mv
DropSubscriptionsMutation drop_subscriptions = 12;
// Combined mutation.
// Currently, it can only be Add & Update, which is for sink into table.
CombinedMutation combined = 100;
}
}
Expand Down
17 changes: 14 additions & 3 deletions src/frontend/src/catalog/catalog_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,11 @@ pub trait CatalogWriter: Send + Sync {
async fn alter_owner(&self, object: alter_owner_request::Object, owner_id: u32) -> Result<()>;

/// Replace the source in the catalog.
async fn alter_source(&self, source: PbSource) -> Result<()>;
async fn alter_source(
&self,
source: PbSource,
replace_streaming_job_plan: Option<PbReplaceStreamingJobPlan>,
) -> Result<()>;

async fn alter_parallelism(
&self,
Expand Down Expand Up @@ -488,8 +492,15 @@ impl CatalogWriter for CatalogWriterImpl {
self.wait_version(version).await
}

async fn alter_source(&self, source: PbSource) -> Result<()> {
let version = self.meta_client.alter_source(source).await?;
async fn alter_source(
&self,
source: PbSource,
replace_streaming_job_plan: Option<PbReplaceStreamingJobPlan>,
) -> Result<()> {
let version = self
.meta_client
.alter_source(source, replace_streaming_job_plan)
.await?;
self.wait_version(version).await
}

Expand Down
3 changes: 2 additions & 1 deletion src/frontend/src/handler/alter_source_column.rs
Original file line number Diff line number Diff line change
Expand Up @@ -121,8 +121,9 @@ pub async fn handle_alter_source_column(
catalog.version += 1;

let catalog_writer = session.catalog_writer()?;
let replace_plan = todo!();
catalog_writer
.alter_source(catalog.to_prost(schema_id, db_id))
.alter_source(catalog.to_prost(schema_id, db_id), replace_plan)
.await?;

Ok(PgResponse::empty_result(StatementType::ALTER_SOURCE))
Expand Down
2 changes: 1 addition & 1 deletion src/frontend/src/handler/alter_source_with_sr.rs
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ pub async fn handle_alter_source_with_sr(
pb_source.version += 1;

let catalog_writer = session.catalog_writer()?;
catalog_writer.alter_source(pb_source).await?;
catalog_writer.alter_source(pb_source, todo!()).await?;

Ok(RwPgResponse::empty_result(StatementType::ALTER_SOURCE))
}
Expand Down
6 changes: 5 additions & 1 deletion src/frontend/src/test_utils.rs
Original file line number Diff line number Diff line change
Expand Up @@ -591,7 +591,11 @@ impl CatalogWriter for MockCatalogWriter {
}
}

async fn alter_source(&self, source: PbSource) -> Result<()> {
async fn alter_source(
&self,
source: PbSource,
_replace_streaming_job_plan: Option<PbReplaceStreamingJobPlan>,
) -> Result<()> {
self.catalog.write().update_source(&source);
Ok(())
}
Expand Down
4 changes: 2 additions & 2 deletions src/meta/service/src/ddl_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -677,10 +677,10 @@ impl DdlService for DdlServiceImpl {
&self,
request: Request<AlterSourceRequest>,
) -> Result<Response<AlterSourceResponse>, Status> {
let AlterSourceRequest { source } = request.into_inner();
let AlterSourceRequest { source, plan } = request.into_inner();
let version = self
.ddl_controller
.run_command(DdlCommand::AlterSourceColumn(source.unwrap()))
.run_command(DdlCommand::AlterSource(source.unwrap(), plan))
.await?;
Ok(Response::new(AlterSourceResponse {
status: None,
Expand Down
6 changes: 3 additions & 3 deletions src/meta/src/barrier/command.rs
Original file line number Diff line number Diff line change
Expand Up @@ -211,8 +211,8 @@ pub enum CreateStreamingJobType {
}

/// [`Command`] is the input of [`crate::barrier::worker::GlobalBarrierWorker`]. For different commands,
/// it will build different barriers to send, and may do different stuffs after the barrier is
/// collected.
/// it will [build different barriers to send](Self::to_mutation),
/// and may [do different stuffs after the barrier is collected](CommandContext::post_collect).
#[derive(Debug, strum::Display)]
pub enum Command {
/// `Flush` command will generate a checkpoint barrier. After the barrier is collected and committed
Expand Down Expand Up @@ -593,6 +593,7 @@ impl Command {

Command::Pause(_) => {
// Only pause when the cluster is not already paused.
// XXX: what if pause(r1) - pause(r2) - resume(r1) - resume(r2)??
if current_paused_reason.is_none() {
Some(Mutation::Pause(PauseMutation {}))
} else {
Expand Down Expand Up @@ -697,7 +698,6 @@ impl Command {
..
}) = job_type
{
// TODO: support in v2.
let update = Self::generate_update_mutation_for_replace_table(
old_fragments,
merge_updates,
Expand Down
3 changes: 2 additions & 1 deletion src/meta/src/controller/streaming_job.rs
Original file line number Diff line number Diff line change
Expand Up @@ -408,6 +408,7 @@ impl CatalogController {
// TODO: In this function, we also update the `Table` model in the meta store.
// Given that we've ensured the tables inside `TableFragments` are complete, shall we consider
// making them the source of truth and performing a full replacement for those in the meta store?
/// Insert fragments and actors to meta store. Used both for creating new jobs and replacing jobs.
pub async fn prepare_streaming_job(
&self,
stream_job_fragments: &StreamJobFragments,
Expand Down Expand Up @@ -473,7 +474,7 @@ impl CatalogController {
}

if !for_replace {
// // Update dml fragment id.
// Update dml fragment id.
if let StreamingJob::Table(_, table, ..) = streaming_job {
Table::update(table::ActiveModel {
table_id: Set(table.id as _),
Expand Down
Loading

0 comments on commit 87e92dd

Please sign in to comment.