public static Set<String> findReferencedColumn(String columnName, ResolvedSchema schema) {
        Column column =
                schema.getColumn(columnName)
                        .orElseThrow(
                                () ->
                                        new ValidationException(
                                                String.format(
                                                        "The input column %s doesn't exist in the schema.",
                                                        columnName)));
        if (!(column instanceof Column.ComputedColumn)) {
            return Collections.emptySet();
        }
        ColumnReferenceVisitor visitor =
                new ColumnReferenceVisitor(
                        // the input ref index is based on a projection of non-computed columns
                        schema.getColumns().stream()
                                .filter(c -> !(c instanceof Column.ComputedColumn))
                                .map(Column::getName)
                                .collect(Collectors.toList()));
        return visitor.visit(((Column.ComputedColumn) column).getExpression());
    }
@Test
    void testFindReferencedColumn() {
        assertThat(ColumnReferenceFinder.findReferencedColumn("b", resolvedSchema))
                .isEqualTo(Collections.emptySet());

        assertThat(ColumnReferenceFinder.findReferencedColumn("a", resolvedSchema))
                .containsExactlyInAnyOrder("b");

        assertThat(ColumnReferenceFinder.findReferencedColumn("c", resolvedSchema))
                .containsExactlyInAnyOrder("d", "e");

        assertThat(ColumnReferenceFinder.findReferencedColumn("ts", resolvedSchema))
                .containsExactlyInAnyOrder("tuple");

        assertThat(ColumnReferenceFinder.findWatermarkReferencedColumn(resolvedSchema))
                .containsExactlyInAnyOrder("ts", "g");
    }