Add support for JOIN PUSHDOWN in Exasol connector

Author: skyglass

docs/src/main/sphinx/connector/exasol.md

@@ -206,3 +206,16 @@ FROM
 
 ```{include} query-table-function-ordering.fragment
 ```
+
+(exasol-pushdown)=
+### Pushdown
+
+The connector supports pushdown for the following operations:
+
+- {ref}`join-pushdown`
+
+```{include} pushdown-correctness-behavior.fragment
+```
+
+```{include} join-pushdown-enabled-true.fragment
+```

plugin/trino-exasol/src/main/java/io/trino/plugin/exasol/ExasolClient.java

@@ -245,8 +245,7 @@ protected void renameTable(ConnectorSession session, Connection connection, Stri
     @Override
     protected boolean isSupportedJoinCondition(ConnectorSession session, JdbcJoinCondition joinCondition)
     {
-        // Deactivated because test 'testJoinPushdown()' requires write access which is not implemented for Exasol
-        return false;
+        return true;
     }
 
     @Override

plugin/trino-exasol/src/test/java/io/trino/plugin/exasol/TestExasolConnectorTest.java

@@ -23,6 +23,7 @@
 import io.trino.testing.sql.SqlExecutor;
 import io.trino.testing.sql.TestTable;
 import io.trino.testing.sql.TestView;
+import org.intellij.lang.annotations.Language;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.parallel.Isolated;
 
@@ -59,9 +60,9 @@ protected QueryRunner createQueryRunner()
     protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
     {
         return switch (connectorBehavior) {
+            case SUPPORTS_JOIN_PUSHDOWN -> true;
             // Tests requires write access which is not implemented
             case SUPPORTS_AGGREGATION_PUSHDOWN,
-                     SUPPORTS_JOIN_PUSHDOWN,
                      SUPPORTS_LIMIT_PUSHDOWN,
                      SUPPORTS_TOPN_PUSHDOWN -> false;
 
@@ -126,6 +127,20 @@ protected TestTable createTableWithUnsupportedColumn()
                 "(one NUMBER(19), two GEOMETRY, three VARCHAR(10 CHAR))");
     }
 
+    @Override
+    // Override, because read-only Exasol connector does not support CREATE TABLE and INSERT statements
+    protected TestTable newTrinoTable(String namePrefix, @Language("SQL") String tableDefinition, List<String> rowsToInsert)
+    {
+        return new TestTable(exasolServer.getSqlExecutor(), TEST_SCHEMA + "." + namePrefix,
+                normalizeTableDefinition(tableDefinition), rowsToInsert);
+    }
+
+    // Normalize to add test schema prefix to the name of the nation table
+    private String normalizeTableDefinition(String original)
+    {
+        return original.replaceAll("FROM nation", "FROM %s.nation".formatted(TEST_SCHEMA));
+    }
+
     @Test
     @Override
     public void testShowColumns()