Add support for JOIN PUSHDOWN in Exasol connector

skyglass · skyglass · commit 98b0b7e7bdba · 2025-09-04T22:49:03.000+02:00
diff --git a/docs/src/main/sphinx/connector/exasol.md b/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
+```
diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/BaseJdbcConnectorTest.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/BaseJdbcConnectorTest.java
@@ -150,12 +150,12 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
         return switch (connectorBehavior) {
             case SUPPORTS_UPDATE -> true;
             case SUPPORTS_ADD_COLUMN_WITH_POSITION,
-                 SUPPORTS_CREATE_MATERIALIZED_VIEW,
-                 SUPPORTS_CREATE_VIEW,
-                 SUPPORTS_DEFAULT_COLUMN_VALUE,
-                 SUPPORTS_MERGE,
-                 SUPPORTS_PREDICATE_EXPRESSION_PUSHDOWN,
-                 SUPPORTS_ROW_LEVEL_UPDATE -> false;
+                    SUPPORTS_CREATE_MATERIALIZED_VIEW,
+                    SUPPORTS_CREATE_VIEW,
+                    SUPPORTS_DEFAULT_COLUMN_VALUE,
+                    SUPPORTS_MERGE,
+                    SUPPORTS_PREDICATE_EXPRESSION_PUSHDOWN,
+                    SUPPORTS_ROW_LEVEL_UPDATE -> false;
             // Dynamic filters can be pushed down only if predicate push down is supported.
             // It is possible for a connector to have predicate push down support but not push down dynamic filters.
             // TODO default SUPPORTS_DYNAMIC_FILTER_PUSHDOWN to SUPPORTS_PREDICATE_PUSHDOWN
@@ -1275,7 +1275,7 @@ public void testJoinPushdown()
         try (TestTable nationLowercaseTable = newTrinoTable(
                 // If a connector supports Join pushdown, but does not allow CTAS, we need to make the table creation here overridable.
                 "nation_lowercase",
-                "AS SELECT nationkey, lower(name) name, regionkey FROM nation")) {
+                getNationLowerCaseTableDefinition())) {
             for (JoinOperator joinOperator : JoinOperator.values()) {
                 log.info("Testing joinOperator=%s", joinOperator);
 
@@ -1290,12 +1290,7 @@ public void testJoinPushdown()
                         .setSystemProperty("enable_dynamic_filtering", "false")
                         .build();
 
-                List<String> nonEqualities = Stream.concat(
-                                Stream.of(JoinCondition.Operator.values())
-                                        .filter(operator -> operator != JoinCondition.Operator.EQUAL && operator != JoinCondition.Operator.IDENTICAL)
-                                        .map(JoinCondition.Operator::getValue),
-                                Stream.of("IS DISTINCT FROM", "IS NOT DISTINCT FROM"))
-                        .collect(toImmutableList());
+                List<String> nonEqualities = getSupportedJoinConditionNonEqualities();
 
                 // basic case
                 assertThat(query(session, format("SELECT r.name, n.name FROM nation n %s region r ON n.regionkey = r.regionkey", joinOperator))).isFullyPushedDown();
@@ -1409,6 +1404,35 @@ public void testJoinPushdown()
         }
     }
 
+    protected List<String> getSupportedJoinConditionNonEqualitiesWithoutDistinctFrom()
+    {
+        List<String> result = new ArrayList<>();
+        for (JoinCondition.Operator op : JoinCondition.Operator.values()) {
+            if (op != JoinCondition.Operator.EQUAL && op != JoinCondition.Operator.IDENTICAL) {
+                result.add(op.getValue());
+            }
+        }
+        return result;
+    }
+
+    protected List<String> getSupportedJoinConditionNonEqualitiesWithDistinctFrom()
+    {
+        List<String> result = new ArrayList<>(getSupportedJoinConditionNonEqualitiesWithoutDistinctFrom());
+        result.add("IS DISTINCT FROM");
+        result.add("IS NOT DISTINCT FROM");
+        return result;
+    }
+
+    protected List<String> getSupportedJoinConditionNonEqualities()
+    {
+        return getSupportedJoinConditionNonEqualitiesWithDistinctFrom();
+    }
+
+    protected String getNationLowerCaseTableDefinition()
+    {
+        return "AS SELECT nationkey, lower(name) name, regionkey FROM nation";
+    }
+
     @Test
     public void testComplexJoinPushdown()
     {
diff --git a/plugin/trino-exasol/src/main/java/io/trino/plugin/exasol/ExasolClient.java b/plugin/trino-exasol/src/main/java/io/trino/plugin/exasol/ExasolClient.java
@@ -15,6 +15,7 @@
 
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
+import io.trino.plugin.base.expression.ConnectorExpressionRewriter;
 import io.trino.plugin.base.mapping.IdentifierMapping;
 import io.trino.plugin.jdbc.BaseJdbcClient;
 import io.trino.plugin.jdbc.BaseJdbcConfig;
@@ -31,6 +32,9 @@
 import io.trino.plugin.jdbc.QueryBuilder;
 import io.trino.plugin.jdbc.WriteFunction;
 import io.trino.plugin.jdbc.WriteMapping;
+import io.trino.plugin.jdbc.expression.JdbcConnectorExpressionRewriterBuilder;
+import io.trino.plugin.jdbc.expression.ParameterizedExpression;
+import io.trino.plugin.jdbc.expression.RewriteIn;
 import io.trino.plugin.jdbc.logging.RemoteQueryModifier;
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.AggregateFunction;
@@ -39,7 +43,11 @@
 import io.trino.spi.connector.ColumnPosition;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.expression.ConnectorExpression;
+import io.trino.spi.type.CharType;
+import io.trino.spi.type.DecimalType;
 import io.trino.spi.type.Type;
+import io.trino.spi.type.VarcharType;
 
 import java.sql.Connection;
 import java.sql.Date;
@@ -52,19 +60,36 @@
 import java.util.Set;
 
 import static io.trino.plugin.jdbc.StandardColumnMappings.bigintColumnMapping;
+import static io.trino.plugin.jdbc.StandardColumnMappings.bigintWriteFunction;
 import static io.trino.plugin.jdbc.StandardColumnMappings.booleanColumnMapping;
+import static io.trino.plugin.jdbc.StandardColumnMappings.booleanWriteFunction;
+import static io.trino.plugin.jdbc.StandardColumnMappings.charWriteFunction;
 import static io.trino.plugin.jdbc.StandardColumnMappings.decimalColumnMapping;
 import static io.trino.plugin.jdbc.StandardColumnMappings.defaultCharColumnMapping;
 import static io.trino.plugin.jdbc.StandardColumnMappings.defaultVarcharColumnMapping;
 import static io.trino.plugin.jdbc.StandardColumnMappings.doubleColumnMapping;
+import static io.trino.plugin.jdbc.StandardColumnMappings.doubleWriteFunction;
 import static io.trino.plugin.jdbc.StandardColumnMappings.integerColumnMapping;
+import static io.trino.plugin.jdbc.StandardColumnMappings.integerWriteFunction;
+import static io.trino.plugin.jdbc.StandardColumnMappings.longDecimalWriteFunction;
+import static io.trino.plugin.jdbc.StandardColumnMappings.shortDecimalWriteFunction;
 import static io.trino.plugin.jdbc.StandardColumnMappings.smallintColumnMapping;
+import static io.trino.plugin.jdbc.StandardColumnMappings.smallintWriteFunction;
+import static io.trino.plugin.jdbc.StandardColumnMappings.tinyintWriteFunction;
+import static io.trino.plugin.jdbc.StandardColumnMappings.varcharWriteFunction;
 import static io.trino.plugin.jdbc.TypeHandlingJdbcSessionProperties.getUnsupportedTypeHandling;
 import static io.trino.plugin.jdbc.UnsupportedTypeHandling.CONVERT_TO_VARCHAR;
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
 import static io.trino.spi.connector.ConnectorMetadata.MODIFYING_ROWS_MESSAGE;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
 import static io.trino.spi.type.DateType.DATE;
 import static io.trino.spi.type.DecimalType.createDecimalType;
+import static io.trino.spi.type.DoubleType.DOUBLE;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.spi.type.SmallintType.SMALLINT;
+import static io.trino.spi.type.TinyintType.TINYINT;
+import static java.lang.String.format;
 import static java.util.Locale.ENGLISH;
 
 public class ExasolClient
@@ -75,6 +100,9 @@ public class ExasolClient
             .add("SYS")
             .build();
 
+    private final ConnectorExpressionRewriter<ParameterizedExpression> connectorExpressionRewriter;
+    private static final int EXASOL_MAX_SUPPORTED_VARCHAR_SIZE = 20000;
+
     @Inject
     public ExasolClient(
             BaseJdbcConfig config,
@@ -84,6 +112,29 @@ public ExasolClient(
             RemoteQueryModifier queryModifier)
     {
         super("\"", connectionFactory, queryBuilder, config.getJdbcTypesMappedToVarchar(), identifierMapping, queryModifier, false);
+        this.connectorExpressionRewriter = JdbcConnectorExpressionRewriterBuilder.newBuilder()
+                .addStandardRules(this::quoted)
+                .add(new RewriteIn())
+                .withTypeClass("integer_type", ImmutableSet.of("tinyint", "smallint", "integer", "bigint"))
+                .withTypeClass("numeric_type", ImmutableSet.of("tinyint", "smallint", "integer", "bigint", "decimal", "real", "double"))
+                .map("$equal(left, right)").to("left = right")
+                .map("$not_equal(left, right)").to("left <> right")
+                .map("$less_than(left, right)").to("left < right")
+                .map("$less_than_or_equal(left, right)").to("left <= right")
+                .map("$greater_than(left, right)").to("left > right")
+                .map("$greater_than_or_equal(left, right)").to("left >= right")
+                .map("$add(left: integer_type, right: integer_type)").to("left + right")
+                .map("$subtract(left: integer_type, right: integer_type)").to("left - right")
+                .map("$multiply(left: integer_type, right: integer_type)").to("left * right")
+                .map("$divide(left: integer_type, right: integer_type)").to("left / right")
+                .map("$modulus(left: integer_type, right: integer_type)").to("left % right")
+                .map("$negate(value: integer_type)").to("-value")
+                .map("$like(value: varchar, pattern: varchar): boolean").to("value LIKE pattern")
+                .map("$not($is_null(value))").to("value IS NOT NULL")
+                .map("$not(value: boolean)").to("NOT value")
+                .map("$is_null(value)").to("value IS NULL")
+                .map("$nullif(first, second)").to("NULLIF(first, second)")
+                .build();
     }
 
     @Override
@@ -95,6 +146,12 @@ protected boolean filterSchema(String schemaName)
         return super.filterSchema(schemaName);
     }
 
+    @Override
+    public Optional<ParameterizedExpression> convertPredicate(ConnectorSession session, ConnectorExpression expression, Map<String, ColumnHandle> assignments)
+    {
+        return connectorExpressionRewriter.rewrite(session, expression, assignments);
+    }
+
     @Override
     public OptionalLong delete(ConnectorSession session, JdbcTableHandle handle)
     {
@@ -188,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
@@ -265,7 +321,48 @@ private static LongWriteFunction dateWriteFunctionUsingSqlDate()
     @Override
     public WriteMapping toWriteMapping(ConnectorSession session, Type type)
     {
-        throw new TrinoException(NOT_SUPPORTED, "This connector does not support writing");
+        if (type == BOOLEAN) {
+            return WriteMapping.booleanMapping("boolean", booleanWriteFunction());
+        }
+
+        if (type == TINYINT) {
+            return WriteMapping.longMapping("tinyint", tinyintWriteFunction());
+        }
+        if (type == SMALLINT) {
+            return WriteMapping.longMapping("smallint", smallintWriteFunction());
+        }
+        if (type == INTEGER) {
+            return WriteMapping.longMapping("integer", integerWriteFunction());
+        }
+        if (type == BIGINT) {
+            return WriteMapping.longMapping("bigint", bigintWriteFunction());
+        }
+
+        if (type == DOUBLE) {
+            return WriteMapping.doubleMapping("double", doubleWriteFunction());
+        }
+
+        if (type instanceof DecimalType decimalType) {
+            String dataType = format("decimal(%s, %s)", decimalType.getPrecision(), decimalType.getScale());
+            if (decimalType.isShort()) {
+                return WriteMapping.longMapping(dataType, shortDecimalWriteFunction(decimalType));
+            }
+            return WriteMapping.objectMapping(dataType, longDecimalWriteFunction(decimalType));
+        }
+
+        if (type instanceof CharType charType) {
+            return WriteMapping.sliceMapping("char(" + charType.getLength() + ")", charWriteFunction());
+        }
+
+        if (type instanceof VarcharType varcharType) {
+            int length = varcharType.isUnbounded()
+                    ? EXASOL_MAX_SUPPORTED_VARCHAR_SIZE
+                    : Math.min(varcharType.getBoundedLength(), EXASOL_MAX_SUPPORTED_VARCHAR_SIZE);
+            String dataType = "varchar(" + length + ")";
+            return WriteMapping.sliceMapping(dataType, varcharWriteFunction());
+        }
+
+        throw new TrinoException(NOT_SUPPORTED, "Unsupported column type: " + type.getDisplayName());
     }
 
     @Override
diff --git a/plugin/trino-exasol/src/test/java/io/trino/plugin/exasol/TestExasolConnectorTest.java b/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;
 
@@ -38,6 +39,7 @@
 import static java.util.stream.IntStream.range;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assumptions.abort;
 
 @Isolated
 final class TestExasolConnectorTest
@@ -59,27 +61,27 @@ 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;
+                    SUPPORTS_TOPN_PUSHDOWN,
+                    SUPPORTS_LIMIT_PUSHDOWN -> false;
 
             // Parallel writing is not supported due to restrictions of the Exasol JDBC driver.
             case SUPPORTS_ADD_COLUMN,
-                 SUPPORTS_ARRAY,
-                 SUPPORTS_COMMENT_ON_TABLE,
-                 SUPPORTS_CREATE_SCHEMA,
-                 SUPPORTS_CREATE_TABLE,
-                 SUPPORTS_DELETE,
-                 SUPPORTS_INSERT,
-                 SUPPORTS_MAP_TYPE,
-                 SUPPORTS_NEGATIVE_DATE, // min date is 0001-01-01
-                 SUPPORTS_RENAME_COLUMN,
-                 SUPPORTS_RENAME_TABLE,
-                 SUPPORTS_ROW_TYPE,
-                 SUPPORTS_SET_COLUMN_TYPE,
-                 SUPPORTS_UPDATE -> false;
+                     SUPPORTS_ARRAY,
+                     SUPPORTS_COMMENT_ON_TABLE,
+                     SUPPORTS_CREATE_SCHEMA,
+                     SUPPORTS_CREATE_TABLE,
+                     SUPPORTS_DELETE,
+                     SUPPORTS_INSERT,
+                     SUPPORTS_MAP_TYPE,
+                     SUPPORTS_NEGATIVE_DATE, // min date is 0001-01-01
+                     SUPPORTS_RENAME_COLUMN,
+                     SUPPORTS_RENAME_TABLE,
+                     SUPPORTS_ROW_TYPE,
+                     SUPPORTS_SET_COLUMN_TYPE,
+                     SUPPORTS_UPDATE -> false;
 
             default -> super.hasBehavior(connectorBehavior);
         };
@@ -126,6 +128,27 @@ 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, tableDefinition, rowsToInsert);
+    }
+
+    @Override
+    // Override to add test schema prefix to the name of the nation table
+    protected String getNationLowerCaseTableDefinition()
+    {
+        return "AS SELECT nationkey, lower(name) name, regionkey FROM %s.nation".formatted(TEST_SCHEMA);
+    }
+
+    @Override
+    // Override, because Exasol doesn't support "IS DISTINCT FROM" and "IS NOT DISTINCT FROM" inequalities
+    protected List<String> getSupportedJoinConditionNonEqualities()
+    {
+        return getSupportedJoinConditionNonEqualitiesWithoutDistinctFrom();
+    }
+
     @Test
     @Override
     public void testShowColumns()
@@ -327,6 +350,20 @@ void testNativeMultipleInClauses()
         onRemoteDatabase().execute(format("SELECT count(*) FROM %s.orders WHERE %s", TEST_SCHEMA, longInClauses));
     }
 
+    @Override
+    @Test
+    public void testNativeQueryInsertStatementTableExists()
+    {
+        abort("Read-only Exasol connector does not support CREATE TABLE and INSERT statements");
+    }
+
+    @Override
+    @Test
+    public void testNativeQuerySelectFromTestTable()
+    {
+        abort("Read-only Exasol connector does not support CREATE TABLE and INSERT statements");
+    }
+
     private static String getLongInClause(int start, int length)
     {
         String longValues = range(start, start + length)
