Incremental Update 4 at Feldera

-- This snippet has two tables and one view:
--
-- * `Stocks` Table: Each row contains a price for a stock at a given time. 
--   A new price for one of the stocks is inserted every second.
--   We configure the datagen connector so that `price_time` is monotonically 
--   increasing (default setting)
-- * `Orders` Table: Each row represents an order for a stock. A new order is inserted every second.
--   We configure the timestamp in `order_time` to be uniformly distributed to show 
--   the workings of the As-Of join.
-- * `Order_With_Stock_Price` view: Matches each order with the most recent stock price based on the
--   order’s timestamp. The As-Of Join finds the latest `price_time` in the Stocks table that is
--   less than or equal to the `order_time`.

CREATE TABLE Stocks (
    symbol VARCHAR NOT NULL,
    price_time BIGINT NOT NULL,  -- UNIX timestamp
    price DOUBLE NOT NULL
) with (
  'connectors' = '[{
    "transport": {
      "name": "datagen",
      "config": {
        "plan": [{ 
            "limit": 1000,
            "rate": 1,
            "fields": {
                "symbol": { "values": ["AAPL", "GOOGL", "SPY", "NVDA"] },
                "price": { "strategy": "uniform", "range": [100, 10000] }
            }
        }]
      }
    }
  }]'
);

CREATE TABLE Orders (
    order_id BIGINT NOT NULL,
    order_time BIGINT NOT NULL,  -- UNIX timestamp
    symbol VARCHAR NOT NULL,
    quantity INT NOT NULL
) with (
  'connectors' = '[{
    "transport": {
      "name": "datagen",
      "config": {
        "plan": [{ 
            "limit": 1000,
            "rate": 1,
            "fields": {
                "order_time": { "strategy": "uniform", "range": [0, 1000] },
                "symbol": { "values": ["AAPL", "GOOGL", "SPY", "NVDA"] },
                "quantity": { "strategy": "zipf", "range": [1, 10000] }
            }
        }]
      }
    }
  }]'
);

CREATE VIEW Order_With_Stock_Price AS (
SELECT
    orders.order_id,
    orders.order_time,
    orders.symbol,
    orders.quantity,
    stocks.price,
    TIMESTAMPADD(SECOND, orders.order_time, TIMESTAMP '1970-01-01') as order_timestamp
FROM
    Orders
LEFT ASOF JOIN Stocks
    MATCH_CONDITION(Stocks.price_time <= Orders.order_time)
    ON Stocks.symbol = Orders.symbol
);