Possible bug in select_with_sql_cte

[jonjonw]

I'm getting an error with a dataset that uses with/cte. And I'm trying to figure out if it's a bug in my code or sequel.

NoMethodError: undefined method `opts' for [{:id=>4, :lineno=>3}]:Array
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/sql.rb:1342:in `literal_dataset_append'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/sql.rb:1661:in `select_with_sql_cte'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/adapters/shared/postgres.rb:2377:in `select_with_sql_cte'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/sql.rb:1646:in `block in select_with_sql'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/sql.rb:1644:in `each'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/sql.rb:1644:in `select_with_sql'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/sql.rb:242:in `_insert_sql'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/sql.rb:34:in `insert_sql'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/dataset/actions.rb:417:in `insert'
    .gems/ruby-3.2.2/gems/sequel-5.67.0/lib/sequel/adapters/shared/postgres.rb:1788:in `insert'
    lib/import/history/bitfinex_daily_history.rb:121:in `insert_asset_history_daily_dataset'
...

Digging in a bit with a debugger, I see at select_with_sql_cte

p cte
> {:args=>[:id, :lineno], :name=>:added_ds_line, :dataset=>[{:id=>5, :lineno=>3}]}

and in insert_sql

p columns
> [:dt, :indicator_id, :asset_id, :settlement_asset_id, :source_id]
p values
#<Sequel::Postgres::Dataset: "
SELECT \"added_ds_line\".\"id\", \"raw\".\"dt\", 19, 2, 5, 4 
FROM \"import_rbf_daily_history\" AS \"raw\" 
INNER JOIN \"added_ds_line\" ON (\"added_ds_line\".\"lineno\" = \"raw\".\"lineno\") 
UNION ALL (
SELECT \"added_ds_line\".\"id\", \"raw\".\"dt\", 20, 2, 5, 4 
FROM \"import_rbf_daily_history\" AS \"raw\" 
INNER JOIN \"added_ds_line\" ON (\"added_ds_line\".\"lineno\" = \"raw\".\"lineno\"))>

Do these seem correct? My code doesn't make a simple example (I'll post one once I can narrow it down a bit more but here are the relevant bits if it's helpful:

datasource_id = 5

# The dataset for the asset_history_daily_dataset insert.
def insert_asset_history_daily_dataset
  select_ds = asset_indicators.keys
                              .map(&method(:select_for_asset_history_daily_dataset))
                              .inject { |memo, ds| memo.union(ds, all: true, from_self: false) }
  AssetHistoryDaily
    .dataset
    .with(:added_ds_line, insert_ds_lines_dataset, args: %i[id lineno])
    .returning
    .insert(%i[dt indicator_id source_id], select_ds)
end

def insert_ds_lines_dataset
  select_ds = DB[:import_rbf_daily_history].select(:lineno, :content, datasource_id)
  DsLine.dataset
        .insert_conflict(constraint: :uniq_ds_lineno)
        .returning(:id, :lineno)
        .insert(%i[lineno content ds_id], select_ds)
end

# The dataset for the insert_asset_history_daily_dataset select.
# Requires .with(:added_ds_line, insert_ds_lines_dataset) to be added to the dataset.
def select_for_asset_history_daily_dataset(indicator_code)
  DB[Sequel[:import_rbf_daily_history].as(:raw)]
    .join(:added_ds_line, lineno: Sequel[:raw][:lineno])
    .select(Sequel[:added_ds_line][:id],
            Sequel[:raw][:dt],
            asset_indicators[indicator_code].pk,
            history_source.id)
end

and the relevant models:

Sequel::Model.plugin :insert_conflict
Sequel::Model.plugin :singular_table_names
DB.extension :pg_streaming
DB.extension :pg_json
DB.extension :pg_row
Sequel.extension :pg_json_ops
Sequel.extension :pg_row_ops

DB.create_table :import_rbf_daily_history, temp: true do
  Integer :lineno, primary_key: true
  column :unix, :bigint
  Date :dt, null: false
  column :v1, :quantity, null: false
  column :v2, :quantity, null: false
  column :content, :jsonb, null: false
end

class AssetHistoryDaily < Sequel.Model(Sequel[:history][:asset_history_daily])
  many_to_one :source, class: :HistorySource
  many_to_one :indicator, class: :Indicator
end

class DsLine < Sequel.Model(Sequel[:import][:ds_line])
end

[jeremyevans]

I think your bug is insert_ds_lines_dataset does not return a dataset, it returns the value of Dataset#insert (an array of hashes as you are using Dataset#returning). Maybe instead of insert(...), you want with_sql(:insert_sql, ...)

[jonjonw]

Thanks for pointing me in the right direction! I was wondering how sequel decides when to 'collapse' datasets with methods like insert in these kinds of situations. Good to know!