Databricks v2

Provides integration with Databricks using the Databricks SQL Statement Execution API for parameterized queries, stored procedures, Delta table mutations, and polling triggers. The Node runs SQL on a Databricks SQL warehouse against a Unity Catalog catalog and schema.

Revision History

0.0.0.10 - Initial release.

Connection

Use the Connection to configure the Databricks workspace, SQL warehouse, Unity Catalog context, and credentials used by all methods.

Setup Notes

1. Identify the Databricks workspace URL, the SQL warehouse ID, and the Unity Catalog catalog and schema that the Workflow will use.
2. For OAuthClientCredentials, create a Databricks service principal with an OAuth secret, and record its client ID and client secret. The client must be allowed the scopes entered in Scopes, including sql and unity-catalog.
3. Grant the principal CAN USE on the SQL warehouse, and USE CATALOG and USE SCHEMA on the configured catalog and schema.
4. Grant table privileges for the methods used in the Workflow. Query, Listen, and ListenChanges require SELECT. Insert, Update, Delete, and Merge require MODIFY. StoredProcedure requires EXECUTE on the procedure.
5. Create a Databricks Connection in the Flowgear Console and enter WorkspaceUrl, WarehouseId, Catalog, and Schema.
6. Select the AuthenticationType. For OAuthClientCredentials, enter ClientId, ClientSecret, and Scopes. For AccessToken, enter the token in AccessToken.
7. Test the Connection. The test confirms that the SQL warehouse and the configured schema are accessible. It does not create schemas.

If the OAuth token request is refused, for example because a requested scope is not allowed for the client, the method fails. Correct the credentials or the client's allowed scopes before running the Workflow again.

Methods

The Databricks Node exposes methods for running parameterized SQL, calling stored procedures, inserting, updating, deleting, and merging Delta table rows, and two trigger methods that poll for changes. Templates are generated from accessible tables, views, and procedures in the configured Catalog and Schema. Template names omit the configured catalog and schema, while the generated SQL and TableName values remain fully qualified. Refresh templates when a table is renamed, removed, or changed.

Query

Executes one parameterized SQL statement and streams the returned rows. Select Blank Query for a parameterized example, or select a table or view template to generate a query and Items schema for that object.

Table and view templates include a RowLimit custom parameter with a default of 10, bound to LIMIT :RowLimit. Use aliases when a query returns duplicate column names.

StoredProcedure

Executes a Unity Catalog stored procedure by using a generated CALL statement and streams the returned rows. Select the template for the procedure to generate the statement and its CustomParameters.

Stored procedures require a Databricks SQL warehouse that supports Unity Catalog stored procedures, such as Databricks Runtime 17.0 or later. Selecting a template reads the procedure's parameters without executing it. Generated templates bind each input to markers named Input1, Input2, and so on, and label each value with the procedure's parameter name. Complex inputs are supplied as JSON strings, and binary inputs are supplied as base64 text.

Templates support IN parameters only. Procedures with OUT or INOUT parameters cannot be loaded as templates. Multiple result sets and output variables are not returned.

Insert

Inserts connected items into a managed or external Delta table.

Insert groups up to 100 consecutive items with the same Properties into one statement, within a 1 MiB request limit. A change in item Properties starts a new batch. The row counts in each response are totals for the whole batch, repeated on every item in that batch. Use BatchItemCount to identify the batch size, and do not add the repeated totals together.

Update

Updates rows identified by exact key values. Only the supplied non-key fields are changed.

Delete

Deletes rows identified by exact key values.

Merge

Updates the row that matches the supplied key values, or inserts a new row when no match exists.

Listen

A trigger method that polls a query and emits the current result snapshot. Use Listen to watch the result of a query, or use ListenChanges to receive individual row changes from a Delta table. Select Blank Listen or a table or view template to generate a query that selects all declared columns.

With RowsChanged, the first snapshot is recorded as a baseline without emitting an event. Later events are emitted when the result changes, including when it becomes empty. With RowsReturned, every non-empty snapshot is emitted, including the first. Comparison ignores row order and Property order, but detects changes to values, nulls, duplicate rows, and the order of nested arrays.

Polling can miss a change that is made and reversed between polls. Keep WatchQuery narrow for large tables, because each snapshot is read in full for comparison. When you replace a template query with a different column selection, refresh the return schema.

ListenChanges

A trigger method that polls the Change Data Feed of a Delta table and emits one event per table commit that contains inserted, updated, or deleted rows.

Each row in Rows contains the table's declared columns plus _change_type, _commit_version, and _commit_timestamp. _change_type is insert, delete, update_preimage, or update_postimage. An update produces separate before and after rows; their order within a commit does not guarantee that they are paired. Commits without change rows advance the listener without emitting an event.

The listener keeps its position in memory only. When the Workflow restarts without StartingVersion, it starts from the current table version and does not read changes made while it was stopped. To resume, store NextVersion after the Workflow has fully processed an event and supply it as StartingVersion on the next start. There is no built-in checkpoint or acknowledgement, so a retry can process the same version again.

The listener stops instead of skipping data when the Change Data Feed is disabled or unavailable, a requested version is no longer retained, the table identity or schema changes, or a commit exceeds MaxCommitRows. Refresh the template after a schema change. Views and non-Delta tables are not available as ListenChanges templates.

Usage Notes

• Use :Name markers with CustomParameters instead of inserting external values into SQL text. CustomParameters accepts JSON scalar values only. To pass a complex value, supply it as a JSON string and convert it in SQL, for example, with from_json.
• Insert, Update, Delete, and Merge target managed or external Delta tables. Table, column, and KeyFields names are matched exactly and are not trimmed. Do not add spaces around the commas in KeyFields.
• Templates suggest key fields from informational primary-key constraints. When no primary key is available, the template suggests the first eligible column. Databricks does not enforce these constraints, so duplicate target keys can affect several rows. Review the suggested key before use.
• Delete accepts key fields only. Remove all non-key Properties from each input item.
• Generated and identity columns can be used as Update and Delete keys, but cannot be assigned or supplied to Insert or Merge.
• Each item must contain at least one known column. An empty connected stream performs no writes.
• Update, Delete, and Merge execute one statement per item. For all four mutation methods, a failure in the first statement or batch stops the method. For later items, input validation and provider failures produce failed responses with the error in Flowgear.Message, and processing continues. Authentication, connection, and service failures stop the method.
• Successful statements remain committed when a later statement fails. If the outcome of a write cannot be confirmed because of a transport failure, the method fails and the statement is not resubmitted.
• Query results are returned inline and are limited by Databricks to 25 MiB. Oversized or truncated results fail.
• Multi-statement scripts are not supported. Use a separate Node invocation for each statement.

Data types

• Integers, decimals, finite floating-point numbers, booleans, ISO dates and timestamps, and base64-encoded binary values are supported.
• Decimals with a precision greater than 28 are returned as strings to preserve up to 38 digits.
• TIMESTAMP values retain their UTC offset, and TIMESTAMP_NTZ values remain without a time zone.
• Arrays of strings and integers are represented as CSV strings. Quote values that contain commas, quotes, or line breaks, and escape a quote by doubling it. Empty text represents an empty array, and an unquoted \N represents a null element.
• STRUCT columns are represented as objects, and arrays of STRUCT as arrays of objects. MAP<STRING,STRING> columns are represented as arrays of Key and Value objects.
• When a table has a single STRUCT column, its child fields appear at the root of each item.
• VARIANT, geospatial, and interval types are not supported. Queries and tables that use them fail instead of omitting fields.

Stored procedure result schemas

Databricks procedure metadata does not describe the columns a procedure returns. Without a declaration, a StoredProcedure template loads with a blank Items schema. You can then use schema inference to populate it, which runs the procedure with the supplied inputs. Use a test environment and suitable test data when you infer a schema.

To declare result columns, set ProcedureResultSchemas on the Connection to a JSON object. Each key is the exact catalog.schema.specific_name identity shown in the template description, which can differ from the procedure's CALL name. Each value maps a result column name to its Databricks SQL type. Declare result-set columns only, not OUT parameters.

{
  "workspace.default.report_identity": {
    "label": "STRING"
  }
}

Update the declaration when the procedure's results change. Invalid JSON or unsupported SQL types prevent the template from loading.

Examples

Query with external values

Set Query to:

SELECT customer_id, company_name
FROM customers
WHERE customer_id = :CustomerId
LIMIT 10

Set CustomParameters to:

{
  "CustomerId": 123
}

Resume ListenChanges after a restart

Store NextVersion from each event after the Workflow has processed all of its Rows, for example, in a database table that persists across Workflow restarts. When the Workflow starts again, supply the stored value as StartingVersion. The listener replays changes from that version if Databricks still retains them.

See also

• Avoiding SQL Injection Attacks.
• Lazy Evaluation and Streaming.
• Triggers.
• Use Delta Lake change data feed on Databricks.
• table_changes table-valued function.