Is there any way to use embed_schema?

[daviaws]

https://hexdocs.pm/ecto/Ecto.Schema.html#embeds_many/3-examples

It's more of a doubt than an issue I faced.

But my new schema has an attributes jsonb which is a string in ch, which is originally an embed for me.

For a simple JSONB as is I used a strategy like this:

defmodule App.ClickHouse.Jsonb do
  use Ecto.Type

  def type, do: Ecto.ParameterizedType.init(Ch, type: "String")

  def cast(value) do
    {:ok, value}
  end

  def dump(value) do
    Jason.encode(value)
  end

  def load(value) do
    Jason.decode(value)
  end
end

And I declared it like this:

field :attributes, Jsonb

Not sure if I could do it in any other way.

[daviaws]

However... even if the approach above worked for a common insert.

For a stream insert as RepoClickhouse.insert_stream(PositionReplicated, positions, settings: [async_insert: 1]) I got:

** Reason for termination ==
** (DBConnection.ConnectionError) client #PID<0.2610.0> stopped: ** (CaseClauseError) no case clause matching: %{}
    (ch 0.1.14) lib/ch/row_binary.ex:180: Ch.RowBinary.encode/2
    (ch 0.1.14) lib/ch/row_binary.ex:76: Ch.RowBinary._encode_rows/4
    (ch 0.1.14) lib/ch/row_binary.ex:76: Ch.RowBinary._encode_rows/4

The code in question from deps/ch/lib/ch/row_binary.ex:

  def encode(type, str) when type in [:string, :binary] do
    case str do
      _ when is_binary(str) -> [encode(:varint, byte_size(str)) | str]
      _ when is_list(str) -> [encode(:varint, IO.iodata_length(str)) | str]
      nil -> 0
    end
  end

Extra: I put some breakpoints in Jsonb.cast and Jsonb.dump. But the code isn't reached for streams.

Workaround: As I use this only in my fixtures for now, the workaroud is to: Enum.into(attrs, %{attributes: "{}"}) for streams inserts

[ruslandoga]

The idea behind insert_stream was that it would try to do as little as possible to stay performant.

https://github.com/plausible/ecto_ch/blob/987912d250cfcdd22c2968b0dd8bf56f58d6b17a/lib/ecto/adapters/clickhouse.ex#L81

Otherwise it wouldn't be all that different from insert_all

You can do the pre-processing in Stream.map

records = 
  records
  |> Stream.map(fn record -> Map.update!(record, :attributes, &Jason.encode_to_iodata!/1) end)
  # etc.

Repo.insert_stream(Schema, records)

---

Off-topic, but you probably want to use Jason.encode_to_iodata and move out Ecto.ParameterizedType.init(Ch, type: "String") outside of type function call, otherwise it's inited every time. You can also probably just use :string as type:

defmodule App.ClickHouse.Jsonb do
  use Ecto.Type

  def type, do: :string

  def cast(value) do
    {:ok, value}
  end

  def dump(value) do
    Jason.encode_to_iodata(value)
  end

  def load(value) do
    Jason.decode(value)
  end
end

[daviaws]

@ruslandoga I saw you closed this, but there is any example about using embed schemas?

[ruslandoga]

I haven't used embedded schemas with this adapter yet. What have you tried? If you post your code and the errors it got here, maybe we'd be able to figure out how to use them.

[ruslandoga]

Sorry for closing it, I thought your question was about your JSON type.

[daviaws]

Sorry for closing it, I thought your question was about your JSON type.

No problems, you're doing a great job so far.

I will try something and give you samples soon.

[ruslandoga]

@daviaws 👋

Can this issue be closed?

[daviaws]

Well, we dropped the usage of clickhouse... so I won't be needing any support anymore for now.

We were receiving a lot of inconsistent data on queries and we did not achieved so great performance as the benchmarks first tells... Postgresql is doing better for us... I would like to know where we did wrong... but this answer never came and we rolled back the plan.

If you do not see any value on embed_schemas feel free to close this request and any other of mine.

Good luck and thank you for all the support.