Repository: risingwavelabs/risingwave (language: python)

Issue:
Support Avro ref type in source

We added the support for Debezium avro, but not otherwise, which makes no sense.

A little background: In debezium, there are `after` and `before` sharing the same schema.

https://github.com/risingwavelabs/risingwave/blob/8dfae832334ae1dc0585dcae4b3071e9ee6c9b1d/src/connector/src/parser/debezium/avro_parser.rs#L83

BTW, this line is introduced in this huge refactor: https://github.com/risingwavelabs/risingwave/pull/10096, which is too confusing and scary. How does it work before?
fix: detect cycle ref in proto
I hereby agree to the terms of the [RisingWave Labs, Inc. Contributor License Agreement](https://gist.github.com/TennyZhuang/f00be7f16996ea48effb049aa7be4d66#file-rw_cla).

## What's changed and what's your intention?

resolve #10475

<!--

**Please do not leave this empty!**

Please explain **IN DETAIL** what the changes are in this PR and why they are needed:

- Summarize your change (**mandatory**)
- How does this PR work? Need a brief introduction for the changed logic (optional)
- Describe clearly one logical change and avoid lazy messages (optional)
- Describe any limitations of the current code (optional)
- Refer to a related PR or issue link (optional)

-->

## Checklist

- [x] I have written necessary rustdoc comments
- [x] I have added necessary unit tests and integration tests
- [ ] I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features #7934).
- [ ] My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
- [x] All checks passed in `./risedev check` (or alias, `./risedev c`)
- [ ] My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)
<!-- To manually trigger a benchmark, please check out [Notion](https://www.notion.so/risingwave-labs/Manually-trigger-nexmark-performance-dashboard-test-b784f1eae1cf48889b2645d020b6b7d3). -->

## Documentation

- [ ] My PR contains user-facing changes.

<!-- 

You can ignore or delete the section below if your PR does not contain user-facing changes.

Otherwise, please write a release note below.

-->

<details><summary>Click here for Documentation</summary>

### Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

- Installation and deployment
- Connector (sources & sinks)
- SQL commands, functions, and operators
- RisingWave cluster configuration changes
- Other (please specify in the release note below)

### Release note

<!--
Please create a release note for your changes. 

Discuss technical details in the "What's changed" section, and 
focus on the impact on users in the release note.

You should also mention the environment or conditions where the impact may occur.
-->

</details>

Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:
resolve ref in schema is an internal API and has some bugs, we use our patched crate in Cargo.toml. Not sure @xiangjinwu includes the fix when migrating deps. 
> Not sure @xiangjinwu includes the fix when migrating deps.

Not yet.

> resolve ref in schema is an internal API and has some bugs, we use our patched crate in Cargo.toml. 

The bug is because we use the internal API the wrong way, or the API itself is poorly designed and hard to use. In short, we need to replaces almost all references to `Schema` into `ResolvedSchema` and use functions that take `schemata` (vec of schemas) rather than a single schema. I will work with @xxchan on resolving this.
The current cloning hack implementation fails to handle the following situation:
```
{
  "type": "record",
  "name": "Root",
  "fields": [
    {
      "name": "f1",
      "type": {
        "type": "record",
        "name": "Nested",
        "fields": [
          {
            "name": "f1",
            "type": {
              "type": "enum",
              "name": "Case",
              "symbols": ["A", "B", "C"]
            }
          },
          {
            "name": "f2",
            "type": "Case"
          }
        ]
      }
    },
    {
      "name": "f2",
      "type": "Nested"
    }
  ]
}
```

In `ResolvedAvroSchema::resolved_schema` the field `f2.f2` is not resolved and still a `ref`:
<details>

```
Record(
    RecordSchema {
        name: Name {
            name: "Root",
            namespace: None,
        },
        aliases: None,
        doc: None,
        fields: [
            RecordField {
                name: "f1",
                doc: None,
                aliases: None,
                default: None,
                schema: Record(
                    RecordSchema {
                        name: Name {
                            name: "Nested",
                            namespace: None,
                        },
                        aliases: None,
                        doc: None,
                        fields: [
                            RecordField {
                                name: "f1",
                                doc: None,
                                aliases: None,
                                default: None,
                                schema: Enum(
                                    EnumSchema {
                                        name: Name {
                                            name: "Case",
                                            namespace: None,
                                        },
                                        aliases: None,
                                        doc: None,
                                        symbols: [
                                            "A",
                                            "B",
                                            "C",
                                        ],
                                        default: None,
                                        attributes: {},
                                    },
                                ),
                                order: Ascending,
                                position: 0,
                                custom_attributes: {},
                            },
                            RecordField {
                                name: "f2",
                                doc: None,
                                aliases: None,
                                default: None,
                                schema: Enum(
                                    EnumSchema {
                                        name: Name {
                                            name: "Case",
                                            namespace: None,
                                        },
                                        aliases: None,
                                        doc: None,
                                        symbols: [
                                            "A",
                                            "B",
                                            "C",
                                        ],
                                        default: None,
                                        attributes: {},
                                    },
                                ),
                                order: Ascending,
                                position: 1,
                                custom_attributes: {},
                            },
                        ],
                        lookup: {
                            "f1": 0,
                            "f2": 1,
                        },
                        attributes: {},
                    },
                ),
                order: Ascending,
                position: 0,
                custom_attributes: {},
            },
            RecordField {
                name: "f2",
                doc: None,
                aliases: None,
                default: None,
                schema: Record(
                    RecordSchema {
                        name: Name {
                            name: "Nested",
                            namespace: None,
                        },
                        aliases: None,
                        doc: None,
                        fields: [
                            RecordField {
                                name: "f1",
                                doc: None,
                                aliases: None,
                                default: None,
                                schema: Enum(
                                    EnumSchema {
                                        name: Name {
                                            name: "Case",
                                            namespace: None,
                                        },
                                        aliases: None,
                                        doc: None,
                                        symbols: [
                                            "A",
                                            "B",
                                            "C",
                                        ],
                                        default: None,
                                        attributes: {},
                                    },
                                ),
                                order: Ascending,
                                position: 0,
                                custom_attributes: {},
                            },
                            RecordField {
                                name: "f2",
                                doc: None,
                                aliases: None,
                                default: None,
                                schema: Ref {
                                    name: Name {
                                        name: "Case",
                                        namespace: None,
                                    },
                                },
                                order: Ascending,
                                position: 1,
                                custom_attributes: {},
                            },
                        ],
                        lookup: {
                            "f1": 0,
                            "f2": 1,
                        },
                        attributes: {},
                    },
                ),
                order: Ascending,
                position: 1,
                custom_attributes: {},
            },
        ],
        lookup: {
            "f1": 0,
            "f2": 1,
        },
        attributes: {},
    },
)
```
</details>
## [Codecov](https://app.codecov.io/gh/risingwavelabs/risingwave/pull/10499?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs) Report
> Merging [#10499](https://app.codecov.io/gh/risingwavelabs/risingwave/pull/10499?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs) (d439de1) into [main](https://app.codecov.io/gh/risingwavelabs/risingwave/commit/e21226ab5261da6b5b66ccc48bdf07384f8f8f0d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs) (e21226a) will **increase** coverage by `0.00%`.
> The diff coverage is `97.50%`.

```diff
@@           Coverage Diff           @@
##             main   #10499   +/-   ##
=======================================
  Coverage   70.33%   70.34%           
=======================================
  Files        1274     1274           
  Lines      218988   219023   +35     
=======================================
+ Hits       154034   154066   +32     
- Misses      64954    64957    +3     
```

| Flag | Coverage Δ | |
|---|---|---|
| rust | `70.34% <97.50%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://app.codecov.io/gh/risingwavelabs/risingwave/pull/10499?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs) | Coverage Δ | |
|---|---|---|
| [src/connector/src/parser/protobuf/parser.rs](https://app.codecov.io/gh/risingwavelabs/risingwave/pull/10499?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs#diff-c3JjL2Nvbm5lY3Rvci9zcmMvcGFyc2VyL3Byb3RvYnVmL3BhcnNlci5ycw==) | `59.33% <97.50%> (+4.08%)` | :arrow_up: |

... and [2 files with indirect coverage changes](https://app.codecov.io/gh/risingwavelabs/risingwave/pull/10499/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs)

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=risingwavelabs)



Fix patch:
diff --git a/src/connector/codec/src/decoder/avro/schema.rs b/src/connector/codec/src/decoder/avro/schema.rs
index f523147d8175d..d226098f63d47 100644
--- a/src/connector/codec/src/decoder/avro/schema.rs
+++ b/src/connector/codec/src/decoder/avro/schema.rs
@@ -15,7 +15,7 @@
 use std::sync::{Arc, LazyLock};
 
 use anyhow::Context;
-use apache_avro::schema::{DecimalSchema, RecordSchema, ResolvedSchema, Schema};
+use apache_avro::schema::{DecimalSchema, NamesRef, RecordSchema, ResolvedSchema, Schema};
 use apache_avro::AvroResult;
 use itertools::Itertools;
 use risingwave_common::error::NotImplemented;
@@ -76,20 +76,28 @@ impl MapHandling {
     }
 }
 
-/// This function expects resolved schema (no `Ref`).
-/// FIXME: require passing resolved schema here.
+/// This function expects original schema (with `Ref`).
 /// TODO: change `map_handling` to some `Config`, and also unify debezium.
 /// TODO: use `ColumnDesc` in common instead of PB.
 pub fn avro_schema_to_column_descs(
     schema: &Schema,
     map_handling: Option<MapHandling>,
 ) -> anyhow::Result<Vec<ColumnDesc>> {
+    let resolved = ResolvedSchema::try_from(schema)?;
     if let Schema::Record(RecordSchema { fields, .. }) = schema {
         let mut index = 0;
+        let mut ancestor_records: Vec<String> = vec![];
         let fields = fields
             .iter()
             .map(|field| {
-                avro_field_to_column_desc(&field.name, &field.schema, &mut index, map_handling)
+                avro_field_to_column_desc(
+                    &field.name,
+                    &field.schema,
+                    &mut index,
+                    &mut ancestor_records,
+                    resolved.get_names(),
+                    map_handling,
+                )
             })
             .collect::<anyhow::Result<_>>()?;
         Ok(fields)
@@ -105,10 +113,22 @@ fn avro_field_to_column_desc(
     name: &str,
     schema: &Schema,
     index: &mut i32,
+    ancestor_records: &mut Vec<String>,
+    refs: &NamesRef<'_>,
     map_handling: Option<MapHandling>,
 ) -> anyhow::Result<ColumnDesc> {
-    let data_type = avro_type_mapping(schema, map_handling)?;
+    let data_type = avro_type_mapping(schema, ancestor_records, refs, map_handling)?;
     match schema {
+        Schema::Ref { name: ref_name } => {
+            avro_field_to_column_desc(
+                name,
+                refs[ref_name], // `ResolvedSchema::try_from` already handles lookup failure
+                index,
+                ancestor_records,
+                refs,
+                map_handling,
+            )
+        }
         Schema::Record(RecordSchema {
             name: schema_name,
             fields,
@@ -116,7 +136,16 @@ fn avro_field_to_column_desc(
         }) => {
             let vec_column = fields
                 .iter()
-                .map(|f| avro_field_to_column_desc(&f.name, &f.schema, index, map_handling))
+                .map(|f| {
+                    avro_field_to_column_desc(
+                        &f.name,
+                        &f.schema,
+                        index,
+                        ancestor_records,
+                        refs,
+                        map_handling,
+                    )
+                })
                 .collect::<anyhow::Result<_>>()?;
             *index += 1;
             Ok(ColumnDesc {
@@ -146,9 +175,11 @@ fn avro_field_to_column_desc(
     }
 }
 
-/// This function expects resolved schema (no `Ref`).
+/// This function expects original schema (with `Ref`).
 fn avro_type_mapping(
     schema: &Schema,
+    ancestor_records: &mut Vec<String>,
+    refs: &NamesRef<'_>,
     map_handling: Option<MapHandling>,
 ) -> anyhow::Result<DataType> {
     let data_type = match schema {
@@ -190,16 +221,34 @@ fn avro_type_mapping(
                 return Ok(DataType::Decimal);
             }
 
-            StructType::new(
+            let unique_name = name.fullname(None);
+            if ancestor_records.contains(&unique_name) {
+                bail!(
+                    "circular reference detected in Avro schema: {} -> {}",
+                    ancestor_records.join(" -> "),
+                    unique_name
+                );
+            }
+
+            ancestor_records.push(unique_name);
+            let ty = StructType::new(
                 fields
                     .iter()
-                    .map(|f| Ok((&f.name, avro_type_mapping(&f.schema, map_handling)?)))
+                    .map(|f| {
+                        Ok((
+                            &f.name,
+                            avro_type_mapping(&f.schema, ancestor_records, refs, map_handling)?,
+                        ))
+                    })
                     .collect::<anyhow::Result<Vec<_>>>()?,
             )
-            .into()
+            .into();
+            ancestor_records.pop();
+            ty
         }
         Schema::Array(item_schema) => {
-            let item_type = avro_type_mapping(item_schema.as_ref(), map_handling)?;
+            let item_type =
+                avro_type_mapping(item_schema.as_ref(), ancestor_records, refs, map_handling)?;
             DataType::List(Box::new(item_type))
         }
         Schema::Union(union_schema) => {
@@ -219,7 +268,7 @@ fn avro_type_mapping(
                 "Union contains duplicate types: {union_schema:?}",
             );
             match get_nullable_union_inner(union_schema) {
-                Some(inner) => avro_type_mapping(inner, map_handling)?,
+                Some(inner) => avro_type_mapping(inner, ancestor_records, refs, map_handling)?,
                 None => {
                     // Convert the union to a struct, each field of the struct represents a variant of the union.
                     // Refer to https://github.com/risingwavelabs/risingwave/issues/16273#issuecomment-2179761345 to see why it's not perfect.
@@ -232,10 +281,11 @@ fn avro_type_mapping(
                         // null will mean the whole struct is null
                         .filter(|variant| !matches!(variant, &&Schema::Null))
                         .map(|variant| {
-                            avro_type_mapping(variant, map_handling).and_then(|t| {
-                                let name = avro_schema_to_struct_field_name(variant)?;
-                                Ok((name, t))
-                            })
+                            avro_type_mapping(variant, ancestor_records, refs, map_handling)
+                                .and_then(|t| {
+                                    let name = avro_schema_to_struct_field_name(variant)?;
+                                    Ok((name, t))
+                                })
                         })
                         .try_collect::<_, Vec<_>, _>()
                         .context("failed to convert Avro union to struct")?;
@@ -250,7 +300,12 @@ fn avro_type_mapping(
             {
                 DataType::Decimal
             } else {
-                bail_not_implemented!("Avro type: {:?}", schema);
+                avro_type_mapping(
+                    refs[name], // `ResolvedSchema::try_from` already handles lookup failure
+                    ancestor_records,
+                    refs,
+                    map_handling,
+                )?
             }
         }
         Schema::Map(value_schema) => {
@@ -268,8 +323,13 @@ fn avro_type_mapping(
                     }
                 }
                 Some(MapHandling::Map) | None => {
-                    let value = avro_type_mapping(value_schema.as_ref(), map_handling)
-                        .context("failed to convert Avro map type")?;
+                    let value = avro_type_mapping(
+                        value_schema.as_ref(),
+                        ancestor_records,
+                        refs,
+                        map_handling,
+                    )
+                    .context("failed to convert Avro map type")?;
                     DataType::Map(MapType::from_kv(DataType::Varchar, value))
                 }
             }
diff --git a/src/connector/src/parser/avro/parser.rs b/src/connector/src/parser/avro/parser.rs
index ac93ab3e69807..e817267778a50 100644
--- a/src/connector/src/parser/avro/parser.rs
+++ b/src/connector/src/parser/avro/parser.rs
@@ -229,7 +229,7 @@ impl AvroParserConfig {
     }
 
     pub fn map_to_columns(&self) -> ConnectorResult<Vec<ColumnDesc>> {
-        avro_schema_to_column_descs(&self.schema.resolved_schema, self.map_handling)
+        avro_schema_to_column_descs(&self.schema.original_schema, self.map_handling)
             .map_err(Into::into)
     }
 }


Test patch:
diff --git a/e2e_test/source_inline/kafka/avro/ref.slt b/e2e_test/source_inline/kafka/avro/ref.slt
new file mode 100644
index 0000000000000..15f36cf4c0d5d
--- /dev/null
+++ b/e2e_test/source_inline/kafka/avro/ref.slt
@@ -0,0 +1,128 @@
+control substitution on
+
+
+system ok
+rpk topic create avro-ref
+
+
+system ok
+sr_register avro-ref-value AVRO <<EOF
+{
+  "type": "record",
+  "name": "Node",
+  "fields": [
+    {
+      "name": "value",
+      "type": "int"
+    },
+    {
+      "name": "next",
+      "type": ["null", "Node"]
+    }
+  ]
+}
+EOF
+
+
+statement error
+create source s WITH (${RISEDEV_KAFKA_WITH_OPTIONS_COMMON}, topic = 'avro-ref') FORMAT PLAIN ENCODE AVRO (schema.registry = '${RISEDEV_SCHEMA_REGISTRY_URL}');
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: connector error
+  2: circular reference detected in Avro schema: Node -> Node
+
+
+system ok
+curl -X DELETE "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects/avro-ref-value"
+
+
+system ok
+curl -X DELETE "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects/avro-ref-value?permanent=true"
+
+
+system ok
+sr_register avro-ref-value AVRO <<EOF
+{
+  "type": "record",
+  "name": "Root",
+  "fields": [
+    {
+      "name": "foo",
+      "type": {
+        "type": "record",
+        "name": "Seg",
+        "fields": [
+          {
+            "name": "a",
+            "type": {
+              "type": "record",
+              "name": "Point",
+              "fields": [
+                {
+                  "name": "x",
+                  "type": "int"
+                },
+                {
+                  "name": "y",
+                  "type": "int"
+                }
+              ]
+            }
+          },
+          {
+            "name": "b",
+            "type": "Point"
+          }
+        ]
+      }
+    },
+    {
+      "name": "bar",
+      "type": "Seg"
+    }
+  ]
+}
+EOF
+
+
+statement ok
+create source s WITH (${RISEDEV_KAFKA_WITH_OPTIONS_COMMON}, topic = 'avro-ref') FORMAT PLAIN ENCODE AVRO (schema.registry = '${RISEDEV_SCHEMA_REGISTRY_URL}');
+
+
+system ok
+rpk topic produce avro-ref --schema-id=topic <<EOF
+{"foo":{"a":{"x":3,"y":4},"b":{"x":5,"y":6}},"bar":{"a":{"x":6,"y":5},"b":{"x":4,"y":3}}}
+EOF
+
+
+query IIIIIIII
+select
+  (foo).a.x,
+  (foo).a.y,
+  (foo).b.x,
+  (foo).b.y,
+  (bar).a.x,
+  (bar).a.y,
+  (bar).b.x,
+  (bar).b.y
+from s;
+----
+3 4 5 6 6 5 4 3
+
+
+statement ok
+drop source s;
+
+
+system ok
+curl -X DELETE "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects/avro-ref-value"
+
+
+system ok
+curl -X DELETE "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects/avro-ref-value?permanent=true"
+
+
+system ok
+rpk topic delete 'avro-ref'
diff --git a/src/connector/codec/tests/integration_tests/avro.rs b/src/connector/codec/tests/integration_tests/avro.rs
index d916dc1aba426..2221917cc2b05 100644
--- a/src/connector/codec/tests/integration_tests/avro.rs
+++ b/src/connector/codec/tests/integration_tests/avro.rs
@@ -56,7 +56,7 @@ fn avro_schema_str_to_risingwave_schema(
         ResolvedAvroSchema::create(avro_schema.into()).context("failed to resolve Avro schema")?;
 
     let rw_schema =
-        avro_schema_to_column_descs(&resolved_schema.resolved_schema, config.map_handling)
+        avro_schema_to_column_descs(&resolved_schema.original_schema, config.map_handling)
             .context("failed to convert Avro schema to RisingWave schema")?
             .iter()
             .map(ColumnDesc::from)