Feature: Handle Composite Types

[dclausen]

Feature description

I have a project that I would like to use conduit for but it makes use of composite types e.g.

CREATE TYPE public.money_with_currency AS (currency_code char(3), amount numeric(20,8))

When I attempt to start to run a pipeline, I receive:

ERR node stopped error="node pg-to-log:pg stopped with error: source stream was stopped unexpectedly: error reading from source: read plugin error: failed to fetch next record: fetchers exited unexpectedly: fetcher for table \"payment_transactions\" exited: failed to fetch results: failed to extract schemas: failed to extract payload schema for table payment_transactions: field \"amount\" with OID 20272 cannot be resolved" component=lifecycle.Service node_id=pg-to-log:pg stack=[{"file":"/private/tmp/conduit-20250320-4805-o24r34/conduit-0.13.3/pkg/lifecycle/service.go","func":"github.com/conduitio/conduit/pkg/lifecycle.(*Service).runPipeline.func2","line":743},{"file":"/private/tmp/conduit-20250320-4805-o24r34/conduit-0.13.3/pkg/lifecycle/stream/source.go","func":"github.com/conduitio/conduit/pkg/lifecycle/stream.(*SourceNode).Run","line":146},{"file":"/private/tmp/conduit-20250320-4805-o24r34/conduit-0.13.3/pkg/lifecycle/stream/source.go","func":"github.com/conduitio/conduit/pkg/lifecycle/stream.(*SourceNode).Run.func1","line":90}]

Since it's possible to list the composite types that have been created detailed here: https://dba.stackexchange.com/questions/35497/display-user-defined-types-and-their-details

schema_name | obj_name | column_name | data_type | ordinal_position | is_required | description
-------------+---------------------+-------------+--------------+------------------+-------------+-------------
public | money_with_currency | currency | character(3) | 1 | f |
public | money_with_currency | amount | numeric | 2 | f |
public | typed_id | type | text | 1 | f |
public | typed_id | id | bigint | 2 | f |
public | typed_refid | type | text | 1 | f |
public | typed_refid | refid | text | 2 | f |

Is it possible to support composite types by either creating Avro records/maps or at the least serializing them to json or string?

[lovromazgon]

This is an interesting request. Can you tell me more about your use case? Are you using the composite type to store the currency and amount together in a single column? I'm curious how this affects indexing.

The error comes from this line, where we have a static map of OID to Go types coming from pgx. We would need to properly initialize this map when establishing the connection to the actual DB to get custom types. In addition, the question is how do we encode the value. I'm not sure if we are able to parse it into a map[string]any and create a specific Avro type for it or if we'll have to encode the type into JSON. This would need to be further investigated.

[dclausen]

This is an interesting request. Can you tell me more about your use case? Are you using the composite type to store the
currency and amount together in a single column? I'm curious how this affects indexing.

Yes, that's correct. An index on this column is not needed for our use case since these rows represent transactions and we would query based upon another column that is already indexed such as a transaction id.

I'm open to any approach that allows the data to be sync'd.

[lovromazgon]

The core team talked about this issue and we decided to put it on the backlog, as it's of relatively low importance for us right now. However, if you want to give it a stab and implement this feature, we are happy to assist with code reviews.