Arroyo throws "This feature is not implemented" when attempting to insert data from a CTE

[GoldToothRichards]

Bug

The following query runs from the Arroyo UI, and allows you to see a working preview:

WITH trades as (
    SELECT
        extract_json_string(value, '$.source') as source,
        extract_json_string(value, '$.base') as base,
        extract_json_string(value, '$.quote') as quote,
        extract_json_string(value, '$.exchange') as exchange,
        CAST(array_element(extract_json(value, '$.volume_base'), 1) AS DOUBLE) as volume_base,
        CAST(array_element(extract_json(value, '$.volume_quote'), 1) AS DOUBLE) as volume_quote,
        CAST(array_element(extract_json(value, '$.price'), 1) AS DOUBLE) as price,
        extract_json_string(value, '$.timestamp') as 'timestamp'
    FROM coincap_crypto_trades_source
),
prices as (
    SELECT
        TUMBLE(interval '1 second') as window,
        trades.source as source,
        trades.base as base,
        trades.quote as quote,
        trades.exchange as exchange,
        SUM(trades.volume_base) AS total_volume_base,
        SUM(trades.volume_quote) AS total_volume_quote
    FROM trades
    GROUP BY
        window,
        trades.source,
        trades.base,
        trades.quote,
        trades.exchange
)

-- INSERT INTO coincap_crypto_prices_sink
SELECT
    prices.source as source,
    prices.base as base,
    prices.quote as quote,
    prices.exchange as exchange,
    prices.total_volume_base as total_volume_base,
    prices.total_volume_quote as total_volume_quote,
    prices.total_volume_quote / prices.total_volume_base as vwap,
    prices.window as window
FROM prices;

However, if you un-comment the INSERT line, it will throw a validation error saying "This feature is not implemented".

Workaround

By using CREATE VIEW statements instead of WITH statements, you can get a working preview while including the INSERT statement:

CREATE VIEW trades as (
    SELECT
        extract_json_string(value, '$.source') as source,
        extract_json_string(value, '$.base') as base,
        extract_json_string(value, '$.quote') as quote,
        extract_json_string(value, '$.exchange') as exchange,
        CAST(array_element(extract_json(value, '$.volume_base'), 1) AS DOUBLE) as volume_base,
        CAST(array_element(extract_json(value, '$.volume_quote'), 1) AS DOUBLE) as volume_quote,
        CAST(array_element(extract_json(value, '$.price'), 1) AS DOUBLE) as price,
        extract_json_string(value, '$.timestamp') as 'timestamp'
    FROM coincap_crypto_trades_source
);

CREATE VIEW prices as (
    SELECT
        TUMBLE(interval '1 second') as window,
        trades.source as source,
        trades.base as base,
        trades.quote as quote,
        trades.exchange as exchange,
        SUM(trades.volume_base) AS total_volume_base,
        SUM(trades.volume_quote) AS total_volume_quote
    FROM trades
    GROUP BY
        window,
        trades.source,
        trades.base,
        trades.quote,
        trades.exchange
);

INSERT INTO coincap_crypto_prices_sink
SELECT
    prices.source as source,
    prices.base as base,
    prices.quote as quote,
    prices.exchange as exchange,
    prices.total_volume_base as total_volume_base,
    prices.total_volume_quote as total_volume_quote,
    prices.total_volume_quote / prices.total_volume_base as vwap,
    prices.window as window
FROM prices;

[GoldToothRichards]

Unfortunately, my pipeline still failed even after getting the preview working from the UI.

Logs

• Time: 2024-06-10T17:01:35.244357Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-connectors/src/kafka/sink/mod.rs:160:13:
  Not yet implemented: implement committing state
  Location:

  • File: crates/arroyo-connectors/src/kafka/sink/mod.rs
  • Line: 160
  • Column: 13

• Time: 2024-06-10T17:01:35.245233Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 0
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(1), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:35.347260Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:65:
  Called Option::unwrap() on a None value
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 65

• Time: 2024-06-10T17:01:35.347639Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 1
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(3), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:35.449675Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:35.450050Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 2
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(5), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:35.550997Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:35.551382Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 3
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(7), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:35.653633Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:35.653941Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 4
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(9), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:35.755697Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:35.756064Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 5
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(11), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:35.857151Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:35.857532Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 6
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(13), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:35.959450Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:35.959851Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 7
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(15), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:36.061833Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:36.062147Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 8
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(17), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:36.163922Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-worker/src/lib.rs:462:49:
  Called Result::unwrap() on an Err value: PoisonError { .. }
  Location:

  • File: crates/arroyo-worker/src/lib.rs
  • Line: 462
  • Column: 49

• Time: 2024-06-10T17:01:36.164172Z
  Module: arroyo_controller::states::scheduling
  Message: Failed to start execution on worker
  Details:

  • Job ID: job_X1CDLpovL8
  • Worker ID: 114
  • Attempt: 9
  • Error: Status { code: Cancelled, message: "h2 protocol error: http2 error: stream error received: stream no longer needed", source: Some(tonic::transport::Error(Transport, hyper::Error(Http2, Error { kind: Reset(StreamId(19), CANCEL, Remote) }))) }

• Time: 2024-06-10T17:01:36.265623Z
  Module: arroyo_server_common
  Message: Panicked at crates/arroyo-controller/src/states/scheduling.rs:495:21:
  Failed to start execution on workers WorkerId(114)
  Location:

  • File: crates/arroyo-controller/src/states/scheduling.rs
  • Line: 495
  • Column: 21

• Time: 2024-06-10T17:01:36.265787Z
  Module: arroyo_controller::states
  Message: Fatal state error
  Details:

  • Job ID: job_X1CDLpovL8
  • State: Scheduling
  • Error Message: Failed to start cluster for pipeline
  • Error: task 8722 panicked

[GoldToothRichards]

Hey @mwylde. I tested this again from the master branch after the changes from 655. Everything is working as expected on my end now!