Migrate Oracle Aq To Scalardb

Migrate Oracle Stored Procedures & Triggers to AQ + ScalarDB Consumer Skill

Purpose

Convert Oracle triggers and stored procedures into an event-driven architecture using Oracle Advanced Queuing (AQ) as the messaging layer and ScalarDB Java Transaction API as the consumer layer.

Producer side (Oracle SQL): Triggers and stored procedures are replaced with AQ enqueue operations — triggers contain no business logic and simply call enqueue SPs.

Consumer side (Java): Message consumer classes dequeue from AQ using JMS and write to ScalarDB-managed tables using the Java Transaction API.

This skill produces:

1. A complete SQL file (aq_setup.sql) for the Oracle producer side
2. Java consumer files for the ScalarDB consumer side
3. An AQ migration report documenting all conversions

---

Skill Responsibility

This skill is responsible for:

• Analyzing triggers and stored procedures from extracted schema JSON
• Determining which triggers/SPs should be converted to AQ enqueue patterns
• Generating SQL for payload types, queue tables, queues, modified triggers, and enqueue SPs
• Generating Java consumer service classes (dequeue + ScalarDB Transaction API)
• Generating message POJO classes and helper utilities
• Producing an AQ migration report

This skill is NOT responsible for:

• Orchestration or command handling (handled by /oracle-to-scalardb command)
• Schema extraction (handled by Subagent 1)
• Schema report generation (handled by Subagent 2)
• General migration analysis (handled by Subagent 3)
• Direct SP/trigger to Java conversion without AQ (handled by Subagent 5)
• Creating a full consumer application (build files, main class, deployment config)

---

Input Contract

| Input | Type | Required | Description | |-------|------|----------|-------------| | raw_schema_data.json | File | YES | Extracted Oracle schema data (specifically the plsql section) | | oracle_schema_report.md | File | YES | Schema report (for table/column context needed for accurate Key builders) | | aq-migration-strategy-guide.md | File | YES | Reference doc with AQ conversion patterns and code examples | | aq-exception-handling-strategy.md | File | YES | Exception classification and retry/commit strategy for consumer error handling | | output_directory | Directory | YES | Where to write generated files |

---

Output Contract

| Output | Location | Description | |--------|----------|-------------| | AQ Setup SQL | /aq_setup.sql | Complete SQL file: payload types, queues, triggers, enqueue SPs | | Java Consumer Classes | /generated-java/Consumer.java | One consumer per queue/message-type | | Java Message POJOs | /generated-java/Message.java | One POJO per payload type | | Java Helper Utility | /generated-java/AqStructHolder.java | Reusable Oracle STRUCT wrapper for ojdbc11 | | Exception Classifier | /generated-java/ExceptionClassifier.java | Classifies exceptions into RETRIABLE / NONRETRIABLE / UNKNOWNTX_STATE for AQ session handling | | AQ Migration Report | /scalardb_aq_migration_report.md | Report documenting all conversions |

---

How to Parse the JSON

Read raw_schema_data.json and extract these sections from plsql:

| JSON Path | Contains | |-----------|----------| | plsql.procedures | Procedure metadata (name, deterministic, parallel, authid) | | plsql.functions | Function metadata (name, return_type, deterministic) | | plsql.packages | Package metadata (name, authid, spec/body status) | | plsql.triggers | Trigger metadata (name, table, timing, event, status) | | plsql.arguments | Parameters for procedures/functions | | plsql.source | PL/SQL source code (grouped by NAME, TYPE, ordered by LINE) | | plsql.trigger_source | Trigger body source code | | plsql.procedure_ddl | Full DDL for procedures | | plsql.function_ddl | Full DDL for functions | | plsql.trigger_ddl | Full DDL for triggers |

Also read oracle_schema_report.md to extract:

• Table names and their columns with data types
• Primary key definitions (needed to build correct Key.of*() calls)
• Foreign key relationships (needed for multi-table operations)

---

Conversion Decision Logic

Which triggers/SPs get converted to AQ?

Analyze each trigger and stored procedure to determine if it should be converted to AQ:

| Object | Convert to AQ? | Rationale | |--------|---------------|-----------| | Trigger with DML (INSERT/UPDATE/DELETE on another table) | YES | The DML becomes a consumer-side ScalarDB operation | | Trigger that calls an SP doing DML | YES | Both trigger and SP are converted | | Trigger with only validation/defaults (no cross-table DML) | NO | Keep as application-layer validation | | SP that INSERTs/UPDATEs/DELETEs records | YES | The DML moves to the consumer | | SP with only SELECT/computation | NO | Convert to direct Java (Subagent 5 handles) | | SP called by a trigger | YES | Becomes the enqueue SP; trigger just calls it |

Trigger conversion rules

1. If a trigger has NO business logic (just calls an SP): The trigger calls the enqueue SP with appropriate parameters. The business logic lives in the consumer.
2. If a trigger HAS business logic: Extract the logic. The trigger calls an enqueue SP passing all needed data (OLD/NEW values). The consumer Java code implements the business logic.
3. Original triggers are DISABLED — new triggers replace them.
4. Preserve the original trigger structure — do NOT split a single trigger into multiple triggers. If the original trigger fires on multiple events (e.g., UPDATE OF job_id, department_id), the AQ replacement MUST be a single trigger with the same event specification. The replacement trigger should call the enqueue SP once, passing all relevant OLD/NEW values.

Stored procedure conversion rules

1. SPs that do DML: Converted to enqueue SPs (DBMS_AQ.ENQUEUE with ON_COMMIT visibility). The actual DML moves to the consumer.
2. SPs called by triggers: Become enqueue SPs. The trigger just calls the enqueue SP.
3. SPs with only SELECT/computation: Not converted to AQ (handled by Subagent 5 as direct Java).

---

SQL Generation Rules

1. Payload Type

Create one Oracle Object Type per distinct message schema. Include all data the consumer needs:

CREATE OR REPLACE TYPE . AS OBJECT (
    -- Include all columns the consumer needs to write
    -- Include operation_type VARCHAR2(20) as a routing key
        ,
    ...
    operation_type   VARCHAR2(20)   -- routing key: identifies what the consumer should do
);

Naming convention: _change_t (e.g., job_history_change_t)

2. Queue Table

BEGIN
    DBMS_AQADM.CREATE_QUEUE_TABLE(
        queue_table        => '.',
        queue_payload_type => '.'
    );
END;
/

Naming convention: _qt (e.g., job_history_qt)

3. Queue

BEGIN
    DBMS_AQADM.CREATE_QUEUE(
        queue_name  => '.',
        queue_table => '.',
        max_retries => 5,
        retry_delay => 0
    );
END;
/

BEGIN
    DBMS_AQADM.START_QUEUE('.');
END;
/

BEGIN
    DBMS_AQADM.GRANT_QUEUE_PRIVILEGE(
        privilege  => 'ALL',
        queue_name => '.',
        grantee    => ''
    );
END;
/

Naming convention: _queue (e.g., job_history_queue)

4. Enqueue Stored Procedures

CREATE OR REPLACE PROCEDURE .SP_ENQUEUE_ (
    
) AS
    l_enq_opts    DBMS_AQ.ENQUEUE_OPTIONS_T;
    l_msg_props   DBMS_AQ.MESSAGE_PROPERTIES_T;
    l_payload     ;
    l_msgid       RAW(16);
BEGIN
    l_payload := (
         => ,
        ...
        operation_type => ''   -- routing key
    );
    l_enq_opts.visibility := DBMS_AQ.ON_COMMIT;
    DBMS_AQ.ENQUEUE(
        queue_name         => '.',
        enqueue_options    => l_enq_opts,
        message_properties => l_msg_props,
        payload            => l_payload,
        msgid              => l_msgid
    );
END SP_ENQUEUE_;
/

5. Modified Triggers

IMPORTANT: Preserve the original trigger structure. Do NOT split a single trigger into multiple triggers. If the original trigger fires on UPDATE OF col1, col2, the replacement trigger MUST use the same event specification as a single trigger.

-- Disable original trigger
ALTER TRIGGER . DISABLE;

-- New trigger: no business logic, just calls the enqueue SP
-- Preserve the SAME event specification as the original trigger
CREATE OR REPLACE TRIGGER .TRG_AQ_
      ON .
    FOR EACH ROW
BEGIN
    SP_ENQUEUE_(
        p_col1 => :OLD.col1,   -- or :NEW.col1 depending on timing
        p_col2 => :NEW.col2,
        ...
    );
END TRG_AQ_;
/

SQL File Structure

The generated aq_setup.sql MUST be organized in this order:

-- =============================================================================
-- Oracle AQ Setup for  Schema
-- Generated: 
-- =============================================================================

-- Section 1: Payload Type Definitions
-- Section 2: Queue Table Creation
-- Section 3: Queue Creation & Configuration
-- Section 4: Disable Original Triggers
-- Section 5: Enqueue Stored Procedures
-- Section 6: New AQ Triggers
-- Section 7: Verification Queries

Note: Keep the AQ setup SQL minimal — only create the necessary payload types, queues, triggers, and enqueue SPs. Avoid unnecessary idempotent cleanup blocks unless the script is designed to be re-runnable.

---

Java Consumer Generation Rules

Target Java Version: 17

All generated Java files MUST target Java 17. Use Java 17 features where appropriate: var, records, instanceof pattern matching, switch expressions, text blocks, List.of(), String.formatted().

Do NOT use preview features or anything requiring Java 21+.

Required Dependencies (document in report)

The following JAR files are required for AQ consumer functionality:

| Dependency | Source | Notes | |------------|--------|-------| | aqapi.jar | Oracle DB ($ORACLE_HOME/rdbms/jlib/aqapi.jar) | Must be extracted from Oracle DB installation or container | | javax.jms-api-2.0.1.jar | Maven Central or Oracle DB | JMS 2.0 API | | ojdbc11-23.x.jar | Maven Central (com.oracle.database.jdbc:ojdbc11) | Oracle JDBC driver | | scalardb-3.17.x.jar | Maven Central (com.scalar-labs:scalardb) | ScalarDB Core (Transaction API) |

Note: ScalarDB Core is the default component (open source/community edition). Only add ScalarDB Cluster dependencies if the SQL interface is needed. For the Java Transaction API used by these consumers, ScalarDB Core is sufficient.

Note: aqapi.jar is NOT available in Maven Central for most versions. It must be obtained from inside the Oracle DB installation directory and added to the project's libs/ folder.

File Naming

• Consumer classes: Consumer.java (PascalCase)
• Example: job_history_queue → JobHistoryQueueConsumer.java
• Message POJOs: Message.java (PascalCase)
• Example: job_history_change_t → JobHistoryChangeMessage.java
• Helper: AqStructHolder.java (always this name)

IMPORTANT: ALL files listed in the Output Contract MUST actually be written to disk. The Generated File Index in the report MUST match the actual files in generated-java/. Do not reference files that were not written.

AqStructHolder (generate once)

package com.example.scalardb.aq;

import oracle.sql.Datum;
import oracle.sql.ORAData;
import oracle.sql.STRUCT;

/**
 * Wraps Oracle STRUCT for ojdbc11 compatibility.
 * Required because oracle.sql.STRUCT no longer implements ORAData directly in ojdbc11.
 *
 * Java version: 17
 * Generated by ScalarDB AQ Migration Skill
 */
public class AqStructHolder implements ORAData {
    private final Datum datum;

    public AqStructHolder(Datum datum) {
        this.datum = datum;
    }

    public STRUCT asStruct() {
        return (STRUCT) datum;
    }

    @Override
    public Datum toDatum(java.sql.Connection conn) {
        return datum;
    }
}

Message POJO (one per payload type)

package com.example.scalardb.aq;

/**
 * Message POJO for Oracle AQ payload type: 
 * Maps to Oracle Object Type attributes by positional index.
 *
 * Java version: 17
 * Generated by ScalarDB AQ Migration Skill
 */
public class Message {
    private  ;     // [0] 
    ...
    private String operationType;  // [N] operation_type (routing key)

    // Getters and setters for all fields
}

Consumer Service Class (one per queue)

package com.example.scalardb.aq;

import com.scalar.db.api.*;
import com.scalar.db.io.Key;
import com.scalar.db.exception.transaction.*;
import oracle.jms.*;
import oracle.sql.STRUCT;
import javax.jms.*;
import java.math.BigDecimal;
import java.util.*;

/**
 * AQ Consumer for queue: 
 * Dequeues messages and processes them using ScalarDB Java Transaction API.
 *
 * Original triggers/SPs converted:
 *   - 
 *
 * Java version: 17
 * Generated by ScalarDB AQ Migration Skill
 */
public class Consumer {

    private static final String NAMESPACE = "";
    private static final String QUEUE_OWNER = "";
    private static final String QUEUE_NAME = "";
    private static final long RECEIVE_TIMEOUT_MS = 10_000;

    private final DistributedTransactionManager txManager;

    public Consumer(DistributedTransactionManager txManager) {
        this.txManager = txManager;
    }

    /**
     * Processes a single dequeued message using ScalarDB Transaction API.
     * Routes to the appropriate handler based on operation_type.
     *
     * @param message the parsed message POJO
     * @throws TransactionException if ScalarDB transaction fails
     */
    public void processMessage(Message message) throws TransactionException {
        switch (message.getOperationType()) {
            case "" -> handle(message);
            case "" -> handle(message);
            default -> throw new IllegalArgumentException(
                "Unknown operation type: " + message.getOperationType());
        }
    }

    // --- Handler methods (one per operation_type) ---

    private void handle(Message msg) throws TransactionException {
        DistributedTransaction tx = txManager.begin();
        try {
            // ScalarDB write logic — Upsert recommended for idempotency
            var upsert = Upsert.newBuilder()
                .namespace(NAMESPACE)
                .table("")
                .partitionKey(Key.of("", msg.get()))
                // .clusteringKey(...) if applicable
                // .xxxValue("", msg.get()) for each column
                .build();
            tx.upsert(upsert);
            tx.commit();
        } catch (Exception e) {
            tx.abort();
            throw e;
        }
    }

    // --- AQ JMS message parsing ---

    /**
     * Parses an Oracle ADT message (STRUCT) into a message POJO.
     * Attribute positions match the Oracle Object Type definition order.
     */
    public static Message parseMessage(Message jmsMessage) throws Exception {
        var adtMsg = (AQjmsAdtMessage) jmsMessage;
        var holder = (AqStructHolder) adtMsg.getAdtPayload();
        Object[] attrs = holder.asStruct().getAttributes();

        var msg = new Message();
        // Map each positional attribute to the POJO field
        // msg.setField(() attrs[index]);
        return msg;
    }
}

ScalarDB Write Patterns

Upsert (recommended for idempotency):

var upsert = Upsert.newBuilder()
    .namespace(NAMESPACE)
    .table("")
    .partitionKey(Key.of("", value))
    .clusteringKey(Key.of("", value))
    .Value("", value)
    .build();
tx.upsert(upsert);

Using Upsert is recommended because it provides idempotency during message redelivery: if the consumer crashes after the ScalarDB commit but before the AQ session commit, AQ redelivers the message and the Upsert simply overwrites with identical data. However, Upsert is not mandatory — Insert can be used if duplicate handling is managed differently.

Insert (alternative):

var insert =

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [wfukatsu](https://github.com/wfukatsu)
- **Source:** [wfukatsu/nexus-architect](https://github.com/wfukatsu/nexus-architect)
- **License:** MIT

Install and usage instructions live in the source repository linked above.