Enable the CDC service

The RAI Native App creates a Change Data Capture (CDC) service to manage a compute engine responsible for processing changes from Snowflake source tables and views. This guide explains how to enable and disable the CDC service, check its status, and resize its engine when needed.

Before you start, make sure that:

• The RAI Native App is installed in your Snowflake account.

What the CDC service does

The CDC service manages a compute engine that processes the change tracking data consumed by data streams to make changes to source data available to semantic models that depend on the streams’ data sources. This service fully manages the lifecycle of the CDC engine, including creating or resuming it when changes are detected and suspending it after inactivity.

Note

The CDC service requires that your RAI Native App be granted the EXECUTE TASK and EXECUTE MANAGED TASK privileges. These are granted automatically when you install the app.

Enable the CDC service

Requires the cdc_admin application role.

Enable CDC when you want the service to start processing source-data changes again. Use this task when you are turning CDC on for the first time or resuming it after a pause.

SQL

-- Enable the CDC service.
CALL relationalai.app.resume_cdc();

Output

+--------------------------------------------------------------------+
| CDC functionality on the RelationalAI application has been resumed |
+--------------------------------------------------------------------+

Python

from relationalai.config import create_config

session = create_config().get_session()

# Enable the CDC service.
session.sql("CALL relationalai.app.resume_cdc()").collect()

Verify the CDC service is ready

Requires the app_user application role.

Use this task after you enable or resume CDC. It helps you confirm that the service is enabled, that the CDC engine can process changes, and that the background task is running.

Run app.cdc_status(), then check CDC_ENABLED and CDC_ENGINE_STATUS:

SQL

-- Get the CDC service status.
CALL relationalai.app.cdc_status();

Output

+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+
| CDC_ENABLED  | CDC_ENGINE_NAME    | CDC_ENGINE_STATUS | CDC_ENGINE_SIZE  | CDC_TASK_STATUS | CDC_TASK_INFO                                                                                                            |
|--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------|
| TRUE         | CDC_MANAGED_ENGINE | READY             | HIGHMEM_X64_S    | started         | {"createdOn": "2024-10-15 21:58:11.291 -0700", "lastSuspendedOn": null, "lastSuspendedReason": null, "state": "started"} |
+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+

Python

from relationalai.config import create_config

session = create_config().get_session()

# Get the CDC service status.
rows = session.sql("CALL relationalai.app.cdc_status()").collect()
print(rows)

Output

+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+
| CDC_ENABLED  | CDC_ENGINE_NAME    | CDC_ENGINE_STATUS | CDC_ENGINE_SIZE  | CDC_TASK_STATUS | CDC_TASK_INFO                                                                                                            |
|--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------|
| TRUE         | CDC_MANAGED_ENGINE | READY             | HIGHMEM_X64_S    | started         | {"createdOn": "2024-10-15 21:58:11.291 -0700", "lastSuspendedOn": null, "lastSuspendedReason": null, "state": "started"} |
+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+

Note that:

• The CDC service is ready if all of the following are true:
  • CDC_ENABLED is TRUE, which confirms that the service is enabled.
  • CDC_ENGINE_STATUS is READY, which confirms that the CDC engine is running and can process changes.
  • CDC_TASK_STATUS is started, which confirms that the background task that manages the CDC engine is running.
• For more info, the CDC_TASK_INFO column shows the latest task state details, such as when the task was created or last suspended.

View the CDC engine size

Requires the app_user application role.

Run app.cdc_status(), then check CDC_ENGINE_SIZE. Use this when you want to confirm which engine size is active or verify that a resize took effect.

SQL

Run app.cdc_status():

-- Get the CDC service status.
CALL relationalai.app.cdc_status();

Output

+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+
| CDC_ENABLED  | CDC_ENGINE_NAME    | CDC_ENGINE_STATUS | CDC_ENGINE_SIZE  | CDC_TASK_STATUS | CDC_TASK_INFO                                                                                                            |
|--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------|
| TRUE         | CDC_MANAGED_ENGINE | READY             | HIGHMEM_X64_S    | started         | {"createdOn": "2024-10-15 21:58:11.291 -0700", "lastSuspendedOn": null, "lastSuspendedReason": null, "state": "started"} |
+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+

Python

Create a Snowpark session, then run app.cdc_status():

from relationalai.config import create_config

session = create_config().get_session()

# Get the CDC service status.
rows = session.sql("CALL relationalai.app.cdc_status()").collect()
print(rows)

Output

+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+
| CDC_ENABLED  | CDC_ENGINE_NAME    | CDC_ENGINE_STATUS | CDC_ENGINE_SIZE  | CDC_TASK_STATUS | CDC_TASK_INFO                                                                                                            |
|--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------|
| TRUE         | CDC_MANAGED_ENGINE | READY             | HIGHMEM_X64_S    | started         | {"createdOn": "2024-10-15 21:58:11.291 -0700", "lastSuspendedOn": null, "lastSuspendedReason": null, "state": "started"} |
+--------------+--------------------+-------------------+------------------+-----------------+--------------------------------------------------------------------------------------------------------------------------+

Note that:

• The CDC_ENGINE_SIZE column shows the current size of the CDC engine, such as HIGHMEM_X64_S or HIGHMEM_X64_M.

Resize the CDC engine

Requires the cdc_admin application role.

Resize the CDC engine when the default size is not sufficient for your workload, such as when change batches are large or processing latency is too high. The new size takes effect the next time a new processing batch starts.

When the RAI Native App is installed, the CDC engine starts with the HIGHMEM_X64_S size. See Compute Pools for background on the available size options.

SQL

To change the size of the CDC engine, use the app.alter_cdc_engine_size() procedure:

-- Change the size of the CDC engine to HIGHMEM_X64_M.
CALL relationalai.app.alter_cdc_engine_size('HIGHMEM_X64_M');

Output

+--------------------------------------+
| CDC engine size set to HIGHMEM_X64_M |
+--------------------------------------+

Python

To change the size of the CDC engine, create a Snowpark session from your default PyRel connection with create_config(), then call the app.alter_cdc_engine_size() procedure:

from relationalai.config import create_config

session = create_config().get_session()

# Change the size of the CDC engine to HIGHMEM_X64_M.
session.sql("CALL relationalai.app.alter_cdc_engine_size('HIGHMEM_X64_M')").collect()

Note

If a batch of stream changes is already being processed, that batch finishes on the previously configured engine size. The new size is used when the next batch starts, at which point the previous engine is deleted.

Disable the CDC service

Requires the cdc_admin application role.

Disable CDC when you need to pause data stream processing, such as when you are investigating a problem or want to reduce costs during a period of inactivity. When CDC is disabled, the service suspends the CDC engine, which pauses all stream change processing until you re-enable CDC.

Note

Disabling CDC does not delete or suspend data streams. Queries against semantic models will continue to run, but they will return results based on the last successfully processed changes until CDC is resumed and the pending changes are processed.

SQL

To disable the CDC service, use the app.suspend_cdc() procedure:

-- Disable the CDC service.
CALL relationalai.app.suspend_cdc();

Output

+------+
| NULL |
+------+

Python

To disable the CDC service, create a Snowpark session from your default PyRel connection with create_config(), then call the app.suspend_cdc() procedure:

from relationalai.config import create_config

session = create_config().get_session()

# Disable the CDC service.
session.sql("CALL relationalai.app.suspend_cdc()").collect()