Pushdown creates invalid SQLs when combining SELECT DISTINCT with constant columns

[Julian-Marx]

When using SELECT DISTINCT on a query that includes constant columns defined as literals (e.g. 'value' as column) the Spark Redshift connector incorrectly pushes down the query in a way that produces invalid SQL for Amazon Redshift.

Specifically, the pushdown optimization treats constant expressions as grouping keys and includes them in the GROUP BY clause. However, Amazon Redshift does not allow constant values in GROUP BY, which leads to query execution failure.

Minimal Example
The following minimal example creates a SQL which redshift rejects:
CodeSnippet:
`def create_view(name):
(spark.read
.format("io.github.spark_redshift_community.spark.redshift")
.option("url", url)
.option("dbtable", name)
.option("tempdir", tempdir)
.option("aws_iam_role", iam_role)
.load()
.createOrReplaceTempView(name)
)

create_view("minimal_table")
transformation="""
select
distinct
'Kontoinhaber' AS PERSON_ROLLE
FROM minimal_table
"""
df=spark.sql(transformation)
df.count()`

Physical plan
== Physical Plan == RedshiftScan [PERSON_ROLLE#2545], SELECT ( 'Kontoinhaber' ) AS "SQ_1_COL_0" FROM ( SELECT * FROM "PUBLIC"."minimal_table" AS "RCQ_ALIAS" ) AS "SQ_0" GROUP BY 'Kontoinhaber', RedshiftRelation("PUBLIC"."minimal_table")

Environment
The issue occurs in Glue:
Glue-Version: 5.1
Spark-Version: 3.5.6-amzn-1
Connector: spark-redshift_2.12-6.4.2-spark_3.5.jar

Error Message
In Redshift: "non-integer constant in GROUP BY"
In Pyspark:
An error occurred while calling o290.count.
: java.sql.SQLException: Exception thrown in awaitResult:
at io.github.spark_redshift_community.spark.redshift.data.JDBCWrapper.executeInterruptibly(JDBCWrapper.scala:98)
at io.github.spark_redshift_community.spark.redshift.data.JDBCWrapper.executeInterruptibly(JDBCWrapper.scala:277)
at io.github.spark_redshift_community.spark.redshift.RedshiftRelation.unloadDataToS3(RedshiftRelation.scala:369)
at io.github.spark_redshift_community.spark.redshift.RedshiftRelation.$anonfun$buildScanFromSQL$1(RedshiftRelation.scala:287)
at scala.Option.orElse(Option.scala:447)
at io.github.spark_redshift_community.spark.redshift.RedshiftRelation.buildScanFromSQL(RedshiftRelation.scala:287)
at io.github.spark_redshift_community.spark.redshift.pushdown.RedshiftScanExec$$anon$1.call(RedshiftScanExec.scala:53)
at io.github.spark_redshift_community.spark.redshift.pushdown.RedshiftScanExec$$anon$1.call(RedshiftScanExec.scala:49)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: com.amazon.redshift.util.RedshiftException: ERROR: non-integer constant in GROUP BY
at com.amazon.redshift.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2648)
at com.amazon.redshift.core.v3.QueryExecutorImpl.processResultsOnThread(QueryExecutorImpl.java:2295)
at com.amazon.redshift.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:1886)
at com.amazon.redshift.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:1878)
at com.amazon.redshift.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:375)
at com.amazon.redshift.jdbc.RedshiftStatementImpl.executeInternal(RedshiftStatementImpl.java:519)
at com.amazon.redshift.jdbc.RedshiftStatementImpl.execute(RedshiftStatementImpl.java:440)
at com.amazon.redshift.jdbc.RedshiftPreparedStatement.executeWithFlags(RedshiftPreparedStatement.java:202)
at com.amazon.redshift.jdbc.RedshiftPreparedStatement.execute(RedshiftPreparedStatement.java:186)
at io.github.spark_redshift_community.spark.redshift.data.JDBCWrapper.$anonfun$executeInterruptibly$2(JDBCWrapper.scala:277)
at io.github.spark_redshift_community.spark.redshift.data.JDBCWrapper.$anonfun$executeInterruptibly$2$adapted(JDBCWrapper.scala:277)
at io.github.spark_redshift_community.spark.redshift.data.JDBCWrapper.$anonfun$executeInterruptibly$1(JDBCWrapper.scala:92)
at scala.concurrent.Future$.$anonfun$apply$1(Future.scala:659)
at scala.util.Success.$anonfun$map$1(Try.scala:255)
at scala.util.Success.map(Try.scala:213)
at scala.concurrent.Future.$anonfun$map$1(Future.scala:292)
at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)
at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
... 3 more