Data Sharing: Publish and Subscribe

Publish tables from a source database in a source workspace, and subscribe to that publication in a target workspace. Publishing and subscribing take place in different workspaces and return their own resources and states.

Task workflow

  1. In the source workspace, read shareable tables and select the publication scope.

  2. In the source workspace, create a publication to obtain the publication resource and metadata.

  3. In the target workspace, read the subscription list to locate the corresponding subscription resource.

  4. Use the subscription resource to subscribe to the publication, confirming results via the returned status and target database.

Prerequisites

Required item

Role on this page

Source and target workspaces

Resource scopes for publishing and subscribing.

Source database

Read shareable tables and establish the publication.

Table scope

Choose between publishing all tables or specific tables.

Target workspace

Receive publication invitations and create subscriptions.

Publication name, subscription name, and remark

Identify publications and target workspace subscriptions.

The publication scope offers two explicit choices: publish all tables, or publish one or more selected tables. When choosing specific tables, select from the source database’s shareable table list; when choosing all tables, do not provide an individual table list.

Creating a publication only indicates that the source workspace received the creation request; it does not mean the target workspace has subscribed.

Publish and subscribe to data

The following example verifies that a specified table exists in the shareable list, creates a publication, locates the subscription resource in the target workspace, subscribes to the publication, and checks the subscription result.

import moi_product_sdk as sdk

def publish_and_subscribe(
    source_workspace,
    target_workspace,
    source_database_id,
    source_table_id,
    target_workspace_id,
    publish_name,
    subscription_name,
    remark,
):
    source_tables = source_workspace.data_share_source_tables(source_database_id)
    if source_table_id not in {item.id for item in source_tables.list}:
        raise ValueError("source table is not available for publication")
    table_scope = sdk.DataShareScope(
        mode="selected",
        object_ids=[source_table_id],
    )
    publish, created = source_workspace.create_data_share_publish(
        publish_name, source_database_id, table_scope,
        [target_workspace_id], sdk.with_data_share_remark(remark),
    )
    if publish.id != created.id:
        raise RuntimeError("publication handle and result do not match")

    subscriptions = target_workspace.data_share().subscriptions()
    subscription_id = next(
        (item.id for item in subscriptions.list if item.pub_name == created.name),
        "",
    )
    if not subscription_id:
        raise RuntimeError("subscription invitation was not found")
    subscription = target_workspace.data_share_subscription(subscription_id)
    subscribed = subscription.subscribe(subscription_name)
    if subscribed.status != "subscribed":
        raise RuntimeError(f"unexpected subscription status: {subscribed.status}")
    return subscribed

import (
	"context"
	"fmt"

	sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
)

func publishAndSubscribe(
	ctx context.Context,
	sourceWorkspace, targetWorkspace *sdk.WorkspaceHandle,
	sourceDatabaseID, sourceTableID, targetWorkspaceID, publishName, subscriptionName, remark string,
) error {
	sourceTables, err := sourceWorkspace.DataShareSourceTables(ctx, sourceDatabaseID)
	if err != nil {
		return err
	}
	found := false
	for _, table := range sourceTables.GetList() {
		if table.GetId() == sourceTableID {
			found = true
			break
		}
	}
	if !found {
		return fmt.Errorf("source table %q is not available for publication", sourceTableID)
	}

	tableScope := &sdk.DataShareScope{
		Mode:      "selected",
		ObjectIds: []string{sourceTableID},
	}
	publish, created, err := sourceWorkspace.CreateDataSharePublish(
		ctx, publishName, sourceDatabaseID, tableScope,
		[]string{targetWorkspaceID}, sdk.WithDataShareRemark(remark),
	)
	if err != nil {
		return err
	}
	if publish.ID() != created.GetId() {
		return fmt.Errorf("publication handle and result do not match")
	}

	subscriptions, err := targetWorkspace.DataShare().Subscriptions(ctx)
	if err != nil {
		return err
	}
	subscriptionID := ""
	for _, item := range subscriptions.GetList() {
		if item.GetPubName() == created.GetName() {
			subscriptionID = item.GetId()
			break
		}
	}
	if subscriptionID == "" {
		return fmt.Errorf("subscription invitation for %q was not found", created.GetName())
	}
	subscription, err := targetWorkspace.DataShareSubscription(subscriptionID)
	if err != nil {
		return err
	}
	subscribed, err := subscription.Subscribe(ctx, subscriptionName)
	if err != nil {
		return err
	}
	if subscribed.GetStatus() != "subscribed" {
		return fmt.Errorf("unexpected subscription status: %s", subscribed.GetStatus())
	}
	return nil
}

The specified table in the example originates from the source database’s shareable table list. After creation, the returned publication resource and metadata point to the same result; you can use the publication resource to update scopes, target workspaces, or remarks, as well as delete the publication.

Result confirmation

Subscription results contain the current subscription status and target database. When downstream status changes need verification, re-read the subscription list in the target workspace; do not assume request success represents the final state.

Limitations

Data sharing impacts data visibility across workspaces. Before creating, updating, or deleting publications, or subscribing and unsubscribing, verify source databases, table scopes, target workspaces, and privileges. Cross-workspace invitation arrivals and status transitions must be confirmed through the target workspace’s subscription list.

Next steps

Last updated on