Toll Collection Demo

-- VoltDB Schema for Toll Collection System
-- Processes license plate scans in real-time with account charging

-------------- REPLICATED TABLES ------------------------------------------------
-- Reference data that doesn't need partitioning

CREATE TABLE VEHICLE_TYPES (
  vehicle_type      SMALLINT           NOT NULL,
  vehicle_class     VARCHAR(20)        NOT NULL,
  toll_multip       DECIMAL            NOT NULL,
  PRIMARY KEY (vehicle_type)
);

CREATE TABLE TOLL_LOCATIONS (
  toll_loc_id       SMALLINT           NOT NULL,
  toll_loc          VARCHAR(64)        NOT NULL,
  toll_loc_status   TINYINT            NOT NULL,
  base_fare         DECIMAL            NOT NULL,
  latitude          DECIMAL            NOT NULL,
  longitude         DECIMAL            NOT NULL,
  PRIMARY KEY (toll_loc_id)
);

CREATE TABLE APPLICATION_PARAMETERS (
  parameter_name    VARCHAR(30) NOT NULL PRIMARY KEY,
  parameter_value   VARCHAR(40) NOT NULL
);

-------------- PARTITIONED TABLES -----------------------------------------------
-- Tables partitioned for parallel processing

CREATE TABLE KNOWN_VEHICLES (
  plate_num         VARCHAR(20)       NOT NULL,
  account_id        INTEGER           NOT NULL,
  vehicle_type      SMALLINT          NOT NULL,
  active            TINYINT           NOT NULL,
  exempt_status     TINYINT           NOT NULL,
  PRIMARY KEY (plate_num)
);
PARTITION TABLE KNOWN_VEHICLES ON COLUMN plate_num;

CREATE TABLE SCAN_HISTORY (
  scan_id           BIGINT            NOT NULL,
  scan_timestamp    TIMESTAMP         NOT NULL,
  plate_num         VARCHAR(20)       NOT NULL,
  account_id        INTEGER,
  toll_loc          VARCHAR(64)       NOT NULL,
  toll_lane_num     VARCHAR(2)        NOT NULL,
  toll_amount       DECIMAL           NOT NULL,
  toll_reason       VARCHAR(200),
  scan_fee_amount   DECIMAL,
  total_amount      DECIMAL           NOT NULL,
  PRIMARY KEY (plate_num, scan_id)
) USING TTL 3600 SECONDS ON COLUMN scan_timestamp;
PARTITION TABLE SCAN_HISTORY ON COLUMN plate_num;

CREATE TABLE ACCOUNTS (
  account_id        INTEGER           NOT NULL,
  account_status    TINYINT           NOT NULL,
  auto_topup        TINYINT           NOT NULL,
  balance           DECIMAL           NOT NULL,
  PRIMARY KEY (account_id)
);
PARTITION TABLE ACCOUNTS ON COLUMN account_id;

-------------- STREAMS ----------------------------------------------------------
-- Output streams for external systems

CREATE STREAM bill_by_mail_stream
PARTITION ON COLUMN plate_num
EXPORT TO TOPIC bill_by_mail_topic
WITH KEY (scan_id) (
  scan_id           BIGINT            NOT NULL,
  scan_timestamp    TIMESTAMP         NOT NULL,
  plate_num         VARCHAR(20)       NOT NULL,
  toll_loc          VARCHAR(20)       NOT NULL,
  toll_lane_num     VARCHAR(2)        NOT NULL,
  toll_amount       DECIMAL           NOT NULL,
  toll_reason       VARCHAR(20),
  scan_fee_amount   DECIMAL,
  tx_fee_amount     DECIMAL,
  total_amount      DECIMAL           NOT NULL
);

-------------- VIEWS ----------------------------------------------------------

CREATE VIEW highest_grossing_locations (toll_loc, total_toll_amount) AS
SELECT toll_loc, SUM(toll_amount) AS total_toll_amount
FROM SCAN_HISTORY GROUP BY toll_loc;

CREATE VIEW location_scans (toll_loc, total_count) AS
SELECT toll_loc, COUNT(*) AS total_count
FROM SCAN_HISTORY GROUP BY toll_loc;

-------------- STORED PROCEDURES -----------------------------------------------

CREATE PROCEDURE PARTITION ON TABLE scan_history COLUMN plate_num PARAMETER 3
FROM CLASS com.voltdb.tollcollect.procedures.ProcessPlate;

/*
 * Copyright (C) 2025 Volt Active Data Inc.
 * ProcessPlate - Stored procedure for toll collection
 */
package com.voltdb.tollcollect.procedures;

import org.voltdb.SQLStmt;
import org.voltdb.VoltProcedure;
import org.voltdb.VoltTable;
import org.voltdb.VoltTableRow;
import org.voltdb.VoltType;
import java.math.BigDecimal;

public class ProcessPlate extends VoltProcedure {

    // SQL statements for lookups
    public final SQLStmt getTollInfo = new SQLStmt(
        "SELECT base_fare FROM TOLL_LOCATIONS WHERE toll_loc = ? AND toll_loc_status = 1;"
    );

    public final SQLStmt getVehicleMultiplier = new SQLStmt(
        "SELECT toll_multip FROM VEHICLE_TYPES WHERE vehicle_class = ?;"
    );

    public final SQLStmt checkVehicle = new SQLStmt(
        "SELECT account_id, exempt_status, vehicle_type FROM KNOWN_VEHICLES " +
        "WHERE plate_num = ? AND active = 1;"
    );

    // SQL statements for inserts
    public final SQLStmt exportBillByMail = new SQLStmt(
        "INSERT INTO bill_by_mail_stream VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);"
    );

    public final SQLStmt insertScanHistory = new SQLStmt(
        "INSERT INTO SCAN_HISTORY VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);"
    );

    public long run(
            long scanTimestamp,
            String location,
            String lane,
            String plateNum,
            String vehicleClass) throws VoltAbortException {

        final long scanId = getUniqueId();

        // Initialize toll calculation variables
        BigDecimal baseToll;
        BigDecimal vehicleMultiplier;
        BigDecimal scanFeeAmount = new BigDecimal(0);
        BigDecimal tollAmount;
        BigDecimal totalAmount;
        String tollReason;
        int accountId = 0;
        byte exemptStatus;

        // Get base toll for location
        voltQueueSQL(getTollInfo, location);
        VoltTable[] tollResults = voltExecuteSQL();
        if (tollResults[0].getRowCount() == 0) {
            throw new VoltAbortException("Invalid toll location");
        }

        // Get vehicle type multiplier
        voltQueueSQL(getVehicleMultiplier, vehicleClass);
        VoltTable[] vehicleResults = voltExecuteSQL();
        if (vehicleResults[0].getRowCount() == 0) {
            throw new VoltAbortException("Invalid vehicle class");
        }

        baseToll = tollResults[0].fetchRow(0).getDecimalAsBigDecimal("base_fare");
        vehicleMultiplier = vehicleResults[0].fetchRow(0).getDecimalAsBigDecimal("toll_multip");
        tollAmount = baseToll.multiply(vehicleMultiplier);

        // Check if vehicle is known
        voltQueueSQL(checkVehicle, plateNum);
        VoltTable[] vehicleCheckResults = voltExecuteSQL();

        if (vehicleCheckResults[0].getRowCount() > 0) {
            VoltTableRow vehicleRow = vehicleCheckResults[0].fetchRow(0);
            accountId = (int) vehicleRow.getLong("account_id");
            exemptStatus = (byte) vehicleRow.get("exempt_status", VoltType.TINYINT);

            if (exemptStatus == 1) {
                tollAmount = BigDecimal.ZERO;
                totalAmount = BigDecimal.ZERO;
                tollReason = "EXEMPT";
            } else {
                totalAmount = tollAmount;
                tollReason = "STANDARD TOLL (" + vehicleClass + ")";
            }
        } else {
            // Unknown vehicle - send to bill-by-mail
            scanFeeAmount = new BigDecimal(2);
            totalAmount = tollAmount.add(scanFeeAmount);
            tollReason = "UNKNOWN_VEHICLE";

            voltQueueSQL(exportBillByMail,
                scanId, new java.util.Date(scanTimestamp), plateNum, location, lane,
                tollAmount, tollReason, scanFeeAmount, null, tollAmount);
            voltExecuteSQL();
        }

        // Insert into scan history
        voltQueueSQL(insertScanHistory,
            scanId, new java.util.Date(scanTimestamp), plateNum, accountId,
            location, lane, tollAmount, tollReason, scanFeeAmount, totalAmount);

        voltExecuteSQL(true);
        return 0;
    }
}

/*
 * VoltSP Java Pipeline for Toll Collection
 * Generates simulated license plate scans at configurable TPS
 */
package com.voltdb.tollcollect.pipeline;

import org.voltdb.stream.api.ExecutionContext;
import org.voltdb.stream.api.Sources;
import org.voltdb.stream.api.pipeline.VoltPipeline;
import org.voltdb.stream.api.pipeline.VoltStreamBuilder;
import org.voltdb.stream.plugin.volt.api.VoltDBResourceConfigBuilder;
import org.voltdb.stream.plugin.volt.api.VoltProcedureRequest;
import org.voltdb.stream.plugin.volt.api.VoltProcedureSinkConfigBuilder;

public class TollCollectStream implements VoltPipeline {

    @Override
    public void define(VoltStreamBuilder stream) {
        // Get configuration values
        ExecutionContext.ConfigurationContext config =
            stream.getExecutionContext().configurator();
        int tps = config.findByPath("tps").asInt();

        PlateRecordGenerator generator = new PlateRecordGenerator();

        stream
            .withName("Toll Collector Data Stream")
            .configureResource("voltdb", VoltDBResourceConfigBuilder.class)
            // Source: Generate plate records at specified TPS
            .consumeFromSource(
                Sources.generateAtRate(
                    tps,
                    generator::generatePlateRecord
                )
            )
            // Transform: Convert PlateRecord to VoltDB procedure call
            .processWith(
                record -> new VoltProcedureRequest("ProcessPlate",
                    new Object[]{
                        record.scanTimestamp(),
                        record.location(),
                        record.lane(),
                        record.plateNum(),
                        record.vehicleClass()
                    })
            )
            // Sink: Send procedure calls to VoltDB
            .terminateWithSink(
                VoltProcedureSinkConfigBuilder.builder()
                    .withVoltClientResourceName("voltdb")
            );
    }
}

// config.yaml example:
// className: com.voltdb.tollcollect.pipeline.TollCollectStream
// tps: 9000