Materialize: Upsert envelope cause duplicate key column when already exists in value

[juanpecatala]

What version of Materialize are you using?

$ materialized -v
materialized v0.3.0 (c9408416ed580b6faf12c605c4f230e3deafa407)

How did you install Materialize?

APT package

What was the issue?

I create a source from a kafka topic with avro format in key and value using confluent schema registry with upsert envelope.
The value already contains the key column so when i show columns from source the column is duplicated with the same name.
We are unable to select one of them and get the following error:

ERROR:  Column name Code is ambiguous

Is the issue reproducible? If so, please provide reproduction instructions.

Create a kafka topic with the following avro key and value.

{
  "type": "record",
  "name": "Key",
  "namespace": "test.source_topic",
  "fields": [
    {
      "name": "Code",
      "type": "string"
    }
  ],
  "connect.name": "source_topic.Key"
}
{
  "type": "record",
  "name": "Value",
  "namespace": "test.source_topic",
  "fields": [
    {
      "name": "Code",
      "type": "string"
    },
    {
      "name": "Name",
      "type": "string"
    }
  ],
  "connect.name": "source_topic.Value"
}

Create a source from that topic and show columns to see duplicate "Code" column.

gz#53

[umanwizard]

@wangandi has the most context on upsert sources, reassigning.

[wangandi]

Hi @juanpecatala! You should be able to work around this issue by renaming the code columns to disambiguate themselves from each other using a syntax similar to this:
CREATE SOURCE src_name(code_key, code_value, name) FROM KAFKA BROKER ...

See here for more information

[wangandi]

As for a more complete solution, we are still trying to gather user feedback on what preferences on that would be. In my mind, there are at least three ways we can handle this issue:
1) Materialize could throw an error in this case and just require the user to manually come up with names to disambiguate the key version of a column from the value version. We would update documentation accordingly.
2) Materialize could automatically disambiguate the key version of a column from the value version, but we'd need feedback on what users would be consider the most intuitive disambiguation scheme.
3) It seems like when this use case arises, the value version of the column is assumed to have identical values to the key version of the column. The upsert convention could come with a contract that Materialize would assume the value version of the column contains identical data to the key version of the column and automatically filter out the value version of the column.

[juanpecatala]

The second option seems most safe (maybe with a prefix like key_column_name) and the third one more clean and I think the most common scenario.
The first one is not an option for me because generate more work in all cases.

[wangandi]

@juanpecatala So that we have a better idea of how we should support this use case and how common the third option would be desired, may I ask, what is the motivation behind having this particular Kafka stream format?

• Is this the format generated by some commonly used stream generation software?
• Is this a standard recommended by some body?
• Is this format particularly useful for some other downstream task?

[juanpecatala]

This format is generated with the common transformation of Debezium "ExtractNewRecordState".
You can see the format generated in the docs https://debezium.io/documentation/reference/1.1/configuration/event-flattening.html

With this option you need less space in Kafka compared with the full Debezium event and is similar to other source connectors like JDBC.

[wangandi]

There is a separate proposal for disambiguating the key version of a column from the value version of a column, which is like this:

Option 2b: Given the fact that we plan on supporting nested avro types (see #1248) and exposing Kafka metadata in sources (see #1577), we could nest the key's version of the column into a key field.
So to select the value version of the column, you would do SELECT code FROM ..., but to select the key version of the column, you would do SELECT key->code FROM ....

Alternatively, the key field could be nested in a general metadata field, and you would select the key version of the column with SELECT metadata->key->code FROM ...

[juanpecatala]

Both options seems good to me.

[wangandi]

In ENVELOPE NONE and ENVELOPE DEBEZIUM, if the fields in the key schema are a subset of the fields in the value schema, we will infer the columns in the value that are also in the key schema to be primary key columns.

In addition to differentiating the key version of a column from the value version of a column, we probably should adopt the same primary key inference behavior in ENVELOPE UPSERT for consistency but have WITH option to disable the primary key inference behavior in case somebody wants to have different data in the key vs value version of the column.

[tenth10th]

Just wanted to ++ implementing similar behavior for ENVELOPE UPSERT - In our use case, we are using the Avro format, and have been duplicating the Kafka Key fields into the Value (mostly for convenience and internal usage reasons), and are encountering the same issue as a result. It would be great to have this fix for non-Debezium UPSERT as well!

[rjnn]

cc @umanwizard for re-triage

[umanwizard]

@wangandi sorry if this is redundant, but since it's been a while since this was discussed, what are your current thoughts on this issue?

[wangandi]

I think we left off at preferring option 2b.

[umanwizard]

@quodlibetor assigning to you since you're working on adding keys to values.

[umanwizard]

@quodlibetor is this roughly the same as #5863 ?