Install and configure the Openflow Connector for Oracle¶

To install the connector, do the following as a data engineer:

1. Navigate to the Openflow overview page. In the Featured connectors section, select View more connectors.

2. On the Openflow connectors page, find the connector and select Install.

3. In the Select runtime dialog, select your runtime from the Available runtimes drop-down list and click Install.

   Note

   Before you install the connector, ensure that you have created a database and schema in Snowflake for the connector to store ingested data.

4. Authenticate to the deployment with your Snowflake account credentials and select Allow when prompted to allow the runtime application to access your Snowflake account. The connector installation process takes a few minutes to complete.

5. Authenticate to the runtime with your Snowflake account credentials.

The Openflow canvas appears with the connector process group added to it.

The runtime size determines the CPU, memory, and disk available to the connector. The available sizes are Small, Medium, and Large. Choose the size when you create the runtime: you can’t change the size of an existing runtime in place.

Size the runtime based on the sustained workload it needs to handle across all connectors running on it. Sustained means typical steady-state throughput, not peak. Peak load can temporarily increase connector queues and end-to-end replication latency; the workload catches up when the load drops back to the steady-state level.

The following ranges are starting points based on internal benchmarks and production customer data. They aren’t service guarantees. Your fit depends on row size, event distribution, schema width, and source burstiness. Start at the lower bound, measure runtime CPU, memory, queue depth, and end-to-end replication latency in production, then increase from there.

• Light workload (aggregate sustained throughput below approximately 1,000 events per second, fewer than approximately 100 actively changing tables): a Small runtime can host a single low-volume connector. Pack additional connectors on Small only when each source is genuinely light.
• Moderate workload (approximately 1,000 to 5,000 events per second, hundreds of actively changing tables): a Medium runtime, typically running 5 to 8 connectors.
• Heavy workload (approximately 5,000 to 15,000 events per second, hundreds to low thousands of actively changing tables): a Large runtime, typically running 15 or more connectors. If you want a smaller blast radius, split across two Medium runtimes instead.

Running multiple connectors on one runtime¶

You can run multiple CDC connector instances on a single runtime. This is useful for replicating many small databases, for example a multi-tenant SaaS with one database per tenant, or a fleet of operational databases per business unit or region.

When you run multiple CDC connector instances of the same type on one runtime, keep their shared Source and Destination parameter contexts intact and override only the per-connector values in each Ingestion context. For the recommended process, see Run multiple CDC connector instances on one runtime in the configuration section for the connector you are setting up.

Each replicated table can consume two Snowpipe Streaming pipes: one for snapshot replication and one for incremental replication. As you pack more tables onto a runtime, check your account’s Snowpipe Streaming pipe limit and raise it before you approach the cap.

Resize a runtime¶

Runtime size is fixed at creation, so to change size you run the connector on a different runtime. You have two options depending on whether you want to preserve the current replication progress.

If you don’t need to keep the progress of the current connector, the simplest path is to create a new runtime at the size you need and install a new connector instance on it. The new connector starts from scratch: it snapshots all configured tables and then captures ongoing changes from that point. The replication progress of the existing connector is discarded.

To keep the progress of the current connector, for example to avoid re-snapshotting tables that took a long time to snapshot initially, migrate the connector to the new runtime. This reuses the existing destination tables and resumes incremental replication from where it left off.

Run multiple CDC connector instances on one runtime¶

Each CDC connector instance uses Source, Destination, and Ingestion parameter contexts. The Ingestion context inherits from the Source and Destination contexts, so any value you don’t override in Ingestion resolves from the parent context.

Use shared Source and Destination contexts with per-connector overrides in Ingestion¶

Snowflake recommends this setup for running multiple CDC connector instances manually. It’s the same pattern the openflow skill applies automatically at scale.

When you import more than one CDC connector instance of the same type into one runtime, keep the Source and Destination contexts at their default names and let every connector instance inherit from them. In that connector’s Ingestion parameter context, override only the values that differ from the shared defaults: the connection URL, replication slot or server ID, destination database, and table list. Leave shared values, such as the Snowflake role, warehouse, and JDBC driver, in the Source and Destination contexts so every connector instance inherits them.

Use the following process for each additional connector instance:

1. Import the connector instance.
2. Confirm that its Ingestion context inherits from the existing Source and Destination contexts, instead of creating a new set of Source and Destination contexts.
3. In the Ingestion context, override every value that must differ from the shared defaults: at minimum, the source connection identity (for example, the JDBC URL and replication slot or server ID) and the destination database.
4. Confirm that the connector replicates from the correct source into the correct destination before moving on to the next connector instance.

Don’t rename the Source and Destination contexts¶

Don’t rename the Source or Destination parameter contexts to make them unique per connector instance. Renaming these contexts might seem like a simpler way to keep each connector instance visually distinct, but it silently breaks future connector version upgrades.

When Snowflake ships a connector version that adds a new parameter to the Source or Destination context, the upgrade process looks for a context with the connector’s default name to apply that new parameter to. If you renamed the context, the upgrade process can’t find it, and the new parameter isn’t added to the context your connector instances actually use. Your connector instances then silently fall out of sync with the new version, and the Snowflake registry can’t repair this automatically. This risk applies whether you rename the contexts to a single shared alternate name or to a distinct name per connector instance.

The Ingestion context is safe to rename, because the registry creates a fresh Ingestion context for every new connector instance. Renaming it doesn’t affect any other connector instance or any future upgrade. If you want every connector instance to be individually identifiable in the parameter context list, rename only its Ingestion context, for example to the source database name, tenant name, or region.

Avoid unintended inheritance from an earlier connector instance¶

The most common mistake when importing an additional CDC connector instance is unintentionally reusing the previous instance’s Ingestion context, instead of creating a dedicated context. If this happens, the new connector instance uses the earlier instance’s source database, destination database, table list, replication slot, server ID, or XStream configuration, and replicates the wrong data without any error.

After importing each additional connector instance, always confirm that its Ingestion context is new and dedicated to that instance, and not shared with any earlier connector instance. If the import wizard offers an option to inherit existing parameter contexts, confirm that it creates a new Ingestion context rather than reusing an existing one.

A table in FAILED state — for example, due to a missing primary key or unsupported schema change — does not restart automatically. If a table enters a FAILED state or you need to restart replication from scratch, use the following procedure to remove and re-add the table to replication.

1. Remove the table from replication, using one of the following methods:

   • Add the table to the Re-snapshot Table Exclusions parameter to temporarily exclude it from replication. This is convenient when the table is matched by an Included Table Regex that you don’t want to change.
   • In the Ingestion Parameters context, either remove the table from Included Table Names or modify the Included Table Regex so the table is no longer matched.

2. Verify the table has been removed:

   1. In the Openflow runtime canvas, right-click a processor group and choose Controller Services.
   2. In the table listing controller services, locate the Table State Store row, click the three vertical dots on the right side of the row, then choose View State.

   Important

   You must wait until the table’s state is fully removed from this list before proceeding. Do not continue until this configuration change has completed.

3. Clean up the destination: Once the table’s state shows as fully removed, manually DROP the destination table in Snowflake. Note that the connector will not overwrite an existing destination table during the snapshot phase; if the table still exists, replication will fail again. Optionally, the journal table and stream can also be removed if they are no longer needed.

4. Re-add the table by reversing the change you made in the first step: either remove the table from Re-snapshot Table Exclusions, or add it back to Included Table Names or Included Table Regex. The connector then re-snapshots the table.

5. Verify the restart: Check the Table State Store using the instructions given previously. The state of the table should appear with the status NEW, then transition to SNAPSHOT_REPLICATION, and finally INCREMENTAL_REPLICATION.

The connector can filter the data replicated per table to a subset of configured columns. Primary key columns are always included regardless of exclusions.

To apply column filters, set the Column Filter JSON parameter in the Ingestion Parameters context to a JSON array of filter objects, one per table you want to filter.

Columns can be included or excluded by name or by regular expression pattern. You can apply a single condition per table, or combine multiple conditions, with exclusions always taking precedence over inclusions.

Syntax¶

Each object in the array identifies a table and specifies which columns to include or exclude. Because this connector uses three-part fully qualified names (database, schema, and table), each object can include a database or databasePattern field in addition to the schema and table fields.

[
    {
        "database": "<database>" | "databasePattern": "<regex>",
        "schema": "<schema>" | "schemaPattern": "<regex>",
        "table": "<table>" | "tablePattern": "<regex>",
        "included": ["<column>", "<column>"],
        "excluded": ["<column>", "<column>"],
        "includedPattern": "<regex>",
        "excludedPattern": "<regex>"
    }
]

The following rules apply:

• Use database, schema, and table for exact name matching, or databasePattern, schemaPattern, and tablePattern for regex matching. You can’t use both a field and its pattern variant in the same object (for example, schema and schemaPattern can’t both appear).
• At least one of included, excluded, includedPattern, or excludedPattern must be provided.
• When both included and excluded filters are specified, exclusions take precedence.
• When multiple filters match the same table, the last matching filter is used, with exact matches taking precedence over pattern-based filters.
• The value can be an array of objects to apply different filters to different tables.

Examples¶

Include specific columns by name:

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "included": ["account_id", "status", "created_at"]
    }
]

Exclude specific columns by name:

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "excluded": ["internal_note", "debug_flag"]
    }
]

Combine an include pattern with a specific exclusion (for example, include all email columns except admin_email):

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "contacts",
        "includedPattern": ".*_email",
        "excluded": ["admin_email"]
    }
]

Mix a database pattern with an exact schema and table name to apply a filter across databases:

[
    {
        "databasePattern": "prod_.*",
        "schema": "dbo",
        "table": "customers",
        "excluded": ["internal_note"]
    }
]

Pass multiple filter objects to apply different rules to different tables:

[
    {"database": "my_db", "schema": "dbo", "table": "orders", "included": ["account_id", "status"]},
    {"database": "my_db", "schema": "dbo", "table": "customers", "excludedPattern": ".*_internal"}
]

Including and excluding the same column¶

Removing a column from a table’s replicated set (by excluding it or by removing it from the included list) has the same effect on the destination as dropping the column at the source: the connector soft-deletes the column on the destination by renaming it with a suffix (by default, __SNOWFLAKE_DELETED). If you then add the column back to the replicated set and later remove it a second time, replication for the affected table fails because the soft-deleted column name is already taken. To recover, restart replication for the affected table.