fix: Don't treat quoted column names as placeholder variables in SQL (#19339)

## Which issue does this PR close?

Closes #19249

## Rationale for this change

SQL allows for column names to begin with `@`, so long as the identifier
is quoted. For example, `SELECT "@column" FROM table`. Both PostgreSQL
and MySQL allow for this at least. The existing SQL parser treats these
column names as placeholder variables, because it does not check if the
identifier is quoted. This change corrects that behavior.

## What changes are included in this PR?

sql/src/expr/identifier.rs - fix to check if identifiers are quoted
sql/tests/common/mod.rs - new test table
sql/tests/sql_integration.rs - unit test

## Are these changes tested?

Yes, `test_parse_quoted_column_name_with_at_sign` unit test was added to
verify quoted column names beginning with `@` are treated as column
names instead of placeholder variables.

## Are there any user-facing changes?

It's possible users are relying on the existing behavior.

Author: pmallex

datafusion/sql/src/expr/identifier.rs

@@ -37,7 +37,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         planner_context: &mut PlannerContext,
     ) -> Result<Expr> {
         let id_span = id.span;
-        if id.value.starts_with('@') {
+        if id.value.starts_with('@') && id.quote_style.is_none() {
             // TODO: figure out if ScalarVariables should be insensitive.
             let var_names = vec![id.value];
             let field = self
@@ -111,7 +111,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                 .filter_map(|id| Span::try_from_sqlparser_span(id.span)),
         );
 
-        if ids[0].value.starts_with('@') {
+        if ids[0].value.starts_with('@') && ids[0].quote_style.is_none() {
             let var_names: Vec<_> = ids
                 .into_iter()
                 .map(|id| self.ident_normalizer.normalize(id))

datafusion/sql/tests/common/mod.rs

@@ -227,6 +227,11 @@ impl ContextProvider for MockContextProvider {
                     false,
                 ),
             ])),
+            "@quoted_identifier_names_table" => Ok(Schema::new(vec![Field::new(
+                "@column",
+                DataType::UInt32,
+                false,
+            )])),
             _ => plan_err!("No table named: {} found", name.table()),
         };
 
@@ -244,8 +249,11 @@ impl ContextProvider for MockContextProvider {
         self.state.aggregate_functions.get(name).cloned()
     }
 
-    fn get_variable_type(&self, _: &[String]) -> Option<DataType> {
-        unimplemented!()
+    fn get_variable_type(&self, variable_names: &[String]) -> Option<DataType> {
+        match variable_names {
+            [var] if var == "@variable" => Some(DataType::Date32),
+            _ => unimplemented!(),
+        }
     }
 
     fn get_window_meta(&self, name: &str) -> Option<Arc<WindowUDF>> {

datafusion/sql/tests/sql_integration.rs

@@ -4522,6 +4522,43 @@ fn test_parse_escaped_string_literal_value() {
     );
 }
 
+#[test]
+fn test_parse_quoted_column_name_with_at_sign() {
+    let sql = r"SELECT `@column` FROM `@quoted_identifier_names_table`";
+    let plan = logical_plan(sql).unwrap();
+    assert_snapshot!(
+        plan,
+        @r#"
+    Projection: @quoted_identifier_names_table.@column
+      TableScan: @quoted_identifier_names_table
+    "#
+    );
+
+    let sql = r"SELECT `@quoted_identifier_names_table`.`@column` FROM `@quoted_identifier_names_table`";
+    let plan = logical_plan(sql).unwrap();
+    assert_snapshot!(
+        plan,
+        @r#"
+    Projection: @quoted_identifier_names_table.@column
+      TableScan: @quoted_identifier_names_table
+    "#
+    );
+}
+
+#[test]
+fn test_variable_identifier() {
+    let sql = r"SELECT t_date32 FROM test WHERE t_date32 = @variable";
+    let plan = logical_plan(sql).unwrap();
+    assert_snapshot!(
+        plan,
+        @r#"
+    Projection: test.t_date32
+      Filter: test.t_date32 = @variable
+        TableScan: test
+    "#
+    );
+}
+
 #[test]
 fn plan_create_index() {
     let sql =