public static List<ProjectionColumn> generateProjectionColumns(
            String projectionExpression,
            List<Column> columns,
            List<UserDefinedFunctionDescriptor> udfDescriptors) {
        if (isNullOrWhitespaceOnly(projectionExpression)) {
            return new ArrayList<>();
        }
        SqlSelect sqlSelect = parseProjectionExpression(projectionExpression);
        if (sqlSelect.getSelectList().isEmpty()) {
            return new ArrayList<>();
        }
        expandWildcard(sqlSelect, columns);
        RelNode relNode = sqlToRel(columns, sqlSelect, udfDescriptors);
        Map<String, RelDataType> relDataTypeMap =
                relNode.getRowType().getFieldList().stream()
                        .collect(
                                Collectors.toMap(
                                        RelDataTypeField::getName, RelDataTypeField::getType));

        Map<String, DataType> rawDataTypeMap =
                columns.stream().collect(Collectors.toMap(Column::getName, Column::getType));

        Map<String, Boolean> isNotNullMap =
                columns.stream()
                        .collect(
                                Collectors.toMap(
                                        Column::getName, column -> !column.getType().isNullable()));

        List<ProjectionColumn> projectionColumns = new ArrayList<>();

        for (SqlNode sqlNode : sqlSelect.getSelectList()) {
            if (sqlNode instanceof SqlBasicCall) {
                SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
                if (SqlKind.AS.equals(sqlBasicCall.getOperator().kind)) {
                    Optional<SqlNode> transformOptional = Optional.empty();
                    String columnName;
                    List<SqlNode> operandList = sqlBasicCall.getOperandList();
                    if (operandList.size() == 2) {
                        transformOptional = Optional.of(operandList.get(0));
                        SqlNode sqlNode1 = operandList.get(1);
                        if (sqlNode1 instanceof SqlIdentifier) {
                            SqlIdentifier sqlIdentifier = (SqlIdentifier) sqlNode1;
                            columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
                        } else {
                            columnName = null;
                        }
                    } else {
                        columnName = null;
                    }
                    if (isMetadataColumn(columnName)) {
                        continue;
                    }
                    ProjectionColumn projectionColumn =
                            transformOptional.isPresent()
                                    ? ProjectionColumn.of(
                                            columnName,
                                            DataTypeConverter.convertCalciteRelDataTypeToDataType(
                                                    relDataTypeMap.get(columnName)),
                                            transformOptional.get().toString(),
                                            JaninoCompiler.translateSqlNodeToJaninoExpression(
                                                    transformOptional.get(), udfDescriptors),
                                            parseColumnNameList(transformOptional.get()))
                                    : ProjectionColumn.of(
                                            columnName,
                                            DataTypeConverter.convertCalciteRelDataTypeToDataType(
                                                    relDataTypeMap.get(columnName)));
                    boolean hasReplacedDuplicateColumn = false;
                    for (int i = 0; i < projectionColumns.size(); i++) {
                        if (projectionColumns.get(i).getColumnName().equals(columnName)
                                && !projectionColumns.get(i).isValidTransformedProjectionColumn()) {
                            hasReplacedDuplicateColumn = true;
                            projectionColumns.set(i, projectionColumn);
                            break;
                        }
                    }
                    if (!hasReplacedDuplicateColumn) {
                        projectionColumns.add(projectionColumn);
                    }
                } else {
                    throw new ParseException(
                            "Unrecognized projection expression: "
                                    + sqlBasicCall
                                    + ". Should be <EXPR> AS <IDENTIFIER>");
                }
            } else if (sqlNode instanceof SqlIdentifier) {
                SqlIdentifier sqlIdentifier = (SqlIdentifier) sqlNode;
                String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
                DataType columnType;
                if (rawDataTypeMap.containsKey(columnName)) {
                    columnType = rawDataTypeMap.get(columnName);
                } else if (relDataTypeMap.containsKey(columnName)) {
                    columnType =
                            DataTypeConverter.convertCalciteRelDataTypeToDataType(
                                    relDataTypeMap.get(columnName));
                } else {
                    throw new RuntimeException(
                            String.format("Failed to deduce column %s type", columnName));
                }
                if (isMetadataColumn(columnName)) {
                    projectionColumns.add(
                            ProjectionColumn.of(
                                    columnName,
                                    // Metadata columns should never be null
                                    columnType.notNull(),
                                    columnName,
                                    columnName,
                                    Arrays.asList(columnName)));
                } else {
                    // Calcite translated column type doesn't keep nullability.
                    // Appending it manually to circumvent this problem.
                    projectionColumns.add(
                            ProjectionColumn.of(
                                    columnName,
                                    isNotNullMap.get(columnName)
                                            ? columnType.notNull()
                                            : columnType.nullable()));
                }
            } else {
                throw new ParseException("Unrecognized projection: " + sqlNode.toString());
            }
        }
        return projectionColumns;
    }
@Test
    public void testGenerateProjectionColumns() {
        List<Column> testColumns =
                Arrays.asList(
                        Column.physicalColumn("id", DataTypes.INT(), "id"),
                        Column.physicalColumn("name", DataTypes.STRING(), "string"),
                        Column.physicalColumn("age", DataTypes.INT(), "age"),
                        Column.physicalColumn("address", DataTypes.STRING(), "address"),
                        Column.physicalColumn("weight", DataTypes.DOUBLE(), "weight"),
                        Column.physicalColumn("height", DataTypes.DOUBLE(), "height"));

        List<ProjectionColumn> result =
                TransformParser.generateProjectionColumns(
                        "id, upper(name) as name, age + 1 as newage, weight / (height * height) as bmi",
                        testColumns,
                        Collections.emptyList());

        List<String> expected =
                Arrays.asList(
                        "ProjectionColumn{column=`id` INT, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
                        "ProjectionColumn{column=`name` STRING, expression='UPPER(`TB`.`name`)', scriptExpression='upper(name)', originalColumnNames=[name], transformExpressionKey=null}",
                        "ProjectionColumn{column=`newage` INT, expression='`TB`.`age` + 1', scriptExpression='age + 1', originalColumnNames=[age], transformExpressionKey=null}",
                        "ProjectionColumn{column=`bmi` DOUBLE, expression='`TB`.`weight` / (`TB`.`height` * `TB`.`height`)', scriptExpression='weight / height * height', originalColumnNames=[weight, height, height], transformExpressionKey=null}");
        Assertions.assertThat(result.toString()).isEqualTo("[" + String.join(", ", expected) + "]");

        List<ProjectionColumn> metadataResult =
                TransformParser.generateProjectionColumns(
                        "*, __namespace_name__, __schema_name__, __table_name__",
                        testColumns,
                        Collections.emptyList());

        List<String> metadataExpected =
                Arrays.asList(
                        "ProjectionColumn{column=`id` INT, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
                        "ProjectionColumn{column=`name` STRING, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
                        "ProjectionColumn{column=`age` INT, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
                        "ProjectionColumn{column=`address` STRING, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
                        "ProjectionColumn{column=`weight` DOUBLE, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
                        "ProjectionColumn{column=`height` DOUBLE, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
                        "ProjectionColumn{column=`__namespace_name__` STRING NOT NULL, expression='__namespace_name__', scriptExpression='__namespace_name__', originalColumnNames=[__namespace_name__], transformExpressionKey=null}",
                        "ProjectionColumn{column=`__schema_name__` STRING NOT NULL, expression='__schema_name__', scriptExpression='__schema_name__', originalColumnNames=[__schema_name__], transformExpressionKey=null}",
                        "ProjectionColumn{column=`__table_name__` STRING NOT NULL, expression='__table_name__', scriptExpression='__table_name__', originalColumnNames=[__table_name__], transformExpressionKey=null}");
        Assertions.assertThat(metadataResult.toString())
                .isEqualTo("[" + String.join(", ", metadataExpected) + "]");

        // calculated columns must use AS to provide an alias name
        Assertions.assertThatThrownBy(
                        () ->
                                TransformParser.generateProjectionColumns(
                                        "id, 1 + 1", testColumns, Collections.emptyList()))
                .isExactlyInstanceOf(ParseException.class);
    }