Allow to enable case insenstive predicate pushdown in JDBC connectors

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/PredicatePushdownController.java

@@ -22,6 +22,7 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static io.trino.plugin.jdbc.JdbcMetadataSessionProperties.getDomainCompactionThreshold;
+import static io.trino.plugin.jdbc.VarcharPredicatePushdownSessionProperties.isUnsafeVarcharPushdownEnabled;
 import static java.util.Objects.requireNonNull;
 
 public interface PredicatePushdownController
@@ -34,6 +35,10 @@ public interface PredicatePushdownController
         return new DomainPushdownResult(domain, Domain.all(domain.getType()));
     };
 
+    PredicatePushdownController PUSHDOWN_AND_KEEP = (session, domain) -> new DomainPushdownResult(
+            domain.simplify(getDomainCompactionThreshold(session)),
+            domain);
+
     PredicatePushdownController DISABLE_PUSHDOWN = (session, domain) -> new DomainPushdownResult(
             Domain.all(domain.getType()),
             domain);
@@ -43,14 +48,16 @@ public interface PredicatePushdownController
                 domain.getType() instanceof VarcharType || domain.getType() instanceof CharType,
                 "CASE_INSENSITIVE_CHARACTER_PUSHDOWN can be used only for chars and varchars");
 
+        if (isUnsafeVarcharPushdownEnabled(session)) {
+            return FULL_PUSHDOWN.apply(session, domain);
+        }
+
         if (domain.isOnlyNull()) {
             return FULL_PUSHDOWN.apply(session, domain);
         }
 
         if (domain.getValues().isDiscreteSet()) {
-            return new DomainPushdownResult(
-                    domain.simplify(getDomainCompactionThreshold(session)),
-                    domain);
+            return PUSHDOWN_AND_KEEP.apply(session, domain);
         }
 
         // case insensitive predicate pushdown could return incorrect results for operators like `!=`, `<` or `>`

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/VarcharPredicatePushdownModule.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.jdbc;
+
+import com.google.inject.Binder;
+import io.airlift.configuration.AbstractConfigurationAwareModule;
+
+import static io.trino.plugin.jdbc.JdbcModule.bindSessionPropertiesProvider;
+
+public class VarcharPredicatePushdownModule
+        extends AbstractConfigurationAwareModule
+{
+    @Override
+    protected void setup(Binder binder)
+    {
+        binder.bind(VarcharPushdownConfig.class);
+        bindSessionPropertiesProvider(binder, VarcharPredicatePushdownSessionProperties.class);
+    }
+}

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/VarcharPredicatePushdownSessionProperties.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.jdbc;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.plugin.base.session.SessionPropertiesProvider;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.session.PropertyMetadata;
+
+import javax.inject.Inject;
+
+import java.util.List;
+
+import static io.trino.spi.session.PropertyMetadata.booleanProperty;
+
+public class VarcharPredicatePushdownSessionProperties
+        implements SessionPropertiesProvider
+{
+    public static final String UNSAFE_VARCHAR_PUSHDOWN = "unsafe-varchar-pushdown";
+
+    private final List<PropertyMetadata<?>> properties;
+
+    @Inject
+    public VarcharPredicatePushdownSessionProperties(VarcharPushdownConfig config)
+    {
+        properties = ImmutableList.<PropertyMetadata<?>>builder()
+                .add(booleanProperty(
+                        UNSAFE_VARCHAR_PUSHDOWN,
+                        "Enable potentially unsafe SQL pushdown based on char or varchar columns. " +
+                                "Note that it may cause incorrect query results.",
+                        config.isUnsafeVarcharPushdownEnabled(),
+                        false))
+                .build();
+    }
+
+    @Override
+    public List<PropertyMetadata<?>> getSessionProperties()
+    {
+        return properties;
+    }
+
+    public static boolean isUnsafeVarcharPushdownEnabled(ConnectorSession session)
+    {
+        return session.getProperty(UNSAFE_VARCHAR_PUSHDOWN, Boolean.class);
+    }
+}

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/VarcharPushdownConfig.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.jdbc;
+
+import io.airlift.configuration.Config;
+import io.airlift.configuration.ConfigDescription;
+
+public class VarcharPushdownConfig
+{
+    private boolean unsafeVarcharPushdownEnabled;
+
+    public boolean isUnsafeVarcharPushdownEnabled()
+    {
+        return unsafeVarcharPushdownEnabled;
+    }
+
+    @Config("unsafe-varchar-pushdown.enabled")
+    @ConfigDescription(
+            "Enable potentially unsafe SQL pushdown based on char or varchar columns. " +
+                    "Note that it may cause incorrect query results.")
+    public VarcharPushdownConfig setUnsafeVarcharPushdownEnabled(boolean unsafeVarcharPushdownEnabled)
+    {
+        this.unsafeVarcharPushdownEnabled = unsafeVarcharPushdownEnabled;
+        return this;
+    }
+}

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/BaseJdbcConnectorTest.java

@@ -18,6 +18,7 @@
 import io.trino.spi.connector.SortOrder;
 import io.trino.sql.planner.assertions.PlanMatchPattern;
 import io.trino.sql.planner.plan.ExchangeNode;
+import io.trino.sql.planner.plan.FilterNode;
 import io.trino.sql.planner.plan.JoinNode;
 import io.trino.sql.planner.plan.TableScanNode;
 import io.trino.sql.planner.plan.TopNNode;
@@ -27,6 +28,7 @@
 import io.trino.testing.sql.TestTable;
 import org.intellij.lang.annotations.Language;
 import org.testng.SkipException;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import java.util.List;
@@ -35,6 +37,7 @@
 import static com.google.common.base.Verify.verify;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.MoreCollectors.toOptional;
+import static io.trino.plugin.jdbc.VarcharPredicatePushdownSessionProperties.UNSAFE_VARCHAR_PUSHDOWN;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.anyTree;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.exchange;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.node;
@@ -48,6 +51,7 @@
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_PREDICATE_PUSHDOWN_WITH_VARCHAR_EQUALITY;
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_PREDICATE_PUSHDOWN_WITH_VARCHAR_INEQUALITY;
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_TOPN_PUSHDOWN;
+import static io.trino.testing.sql.TestTable.randomTableSuffix;
 import static java.lang.String.format;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -553,4 +557,58 @@ protected Session joinPushdownEnabled(Session session)
                 .setCatalogSessionProperty(session.getCatalog().orElseThrow(), "join_pushdown_enabled", "true")
                 .build();
     }
+
+    @Test(dataProvider = "testCaseInsensitivePredicatePushdownDataProvider")
+    public void testCaseInsensitivePredicatePushdown(String type)
+    {
+        Session unsafeVarcharPushdown = Session.builder(getSession())
+                .setCatalogSessionProperty(getSession().getCatalog().orElseThrow(), UNSAFE_VARCHAR_PUSHDOWN, "true")
+                .build();
+
+        if (hasBehavior(SUPPORTS_PREDICATE_PUSHDOWN_WITH_VARCHAR_INEQUALITY)) {
+            assertQueryFails(unsafeVarcharPushdown, "SELECT 1", "Unknown session property .*.unsafe-varchar-pushdown");
+            return;
+        }
+        String tableName = "test_case_insensitive_predicate_pushdown" + randomTableSuffix();
+        assertQuerySucceeds(format("CREATE TABLE %s (value %s)", tableName, type));
+        assertQuerySucceeds(format("INSERT INTO %s VALUES 'a', 'A', 'b', 'B'", tableName));
+
+        assertThat(query(unsafeVarcharPushdown, format("SELECT value FROM %s WHERE value = 'a'", tableName)))
+                // possibly incorrect results with unexpected 'A'
+                .isFullyPushedDown();
+        assertThat(query(unsafeVarcharPushdown, format("SELECT value FROM %s WHERE value != 'a'", tableName)))
+                // possibly incorrect results with missing 'A'
+                .isFullyPushedDown();
+        assertThat(query(unsafeVarcharPushdown, format("SELECT value FROM %s WHERE value > 'A'", tableName)))
+                // possibly incorrect results with missing 'a'
+                .isFullyPushedDown();
+
+        // Verify that results are correct when session property is not used
+        assertThat(query(format("SELECT value FROM %s WHERE value = 'a'", tableName)))
+                .skippingTypesCheck()
+                .matches("VALUES 'a'");
+        assertThat(query(format("SELECT value FROM %s WHERE value != 'a'", tableName)))
+                .isNotFullyPushedDown(FilterNode.class)
+                .skippingTypesCheck()
+                .matches("VALUES 'A', 'b', 'B'");
+        assertThat(query(format("SELECT value FROM %s WHERE value != 'a'", tableName)))
+                .isNotFullyPushedDown(FilterNode.class)
+                .skippingTypesCheck()
+                .matches("VALUES 'A', 'b', 'B'");
+        assertThat(query(format("SELECT value FROM %s WHERE value > 'A'", tableName)))
+                .isNotFullyPushedDown(FilterNode.class)
+                .skippingTypesCheck()
+                .matches("VALUES 'a', 'B', 'b'");
+
+        assertQuerySucceeds("DROP TABLE " + tableName);
+    }
+
+    @DataProvider
+    public Object[][] testCaseInsensitivePredicatePushdownDataProvider()
+    {
+        return new Object[][] {
+                {"char(1)"},
+                {"varchar(1)"},
+        };
+    }
 }

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestVarcharPushdownConfig.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.jdbc;
+
+import com.google.common.collect.ImmutableMap;
+import org.testng.annotations.Test;
+
+import java.util.Map;
+
+import static io.airlift.configuration.testing.ConfigAssertions.assertFullMapping;
+import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults;
+import static io.airlift.configuration.testing.ConfigAssertions.recordDefaults;
+
+public class TestVarcharPushdownConfig
+{
+    @Test
+    public void testDefaults()
+    {
+        assertRecordedDefaults(recordDefaults(VarcharPushdownConfig.class)
+                .setUnsafeVarcharPushdownEnabled(false));
+    }
+
+    @Test
+    public void testExplicitPropertyMappings()
+    {
+        Map<String, String> properties = new ImmutableMap.Builder<String, String>()
+                .put("unsafe-varchar-pushdown.enabled", "true")
+                .build();
+
+        VarcharPushdownConfig expected = new VarcharPushdownConfig()
+                .setUnsafeVarcharPushdownEnabled(true);
+
+        assertFullMapping(properties, expected);
+    }
+}

plugin/trino-memsql/src/main/java/io/trino/plugin/memsql/MemSqlClientModule.java

@@ -24,6 +24,7 @@
 import io.trino.plugin.jdbc.DriverConnectionFactory;
 import io.trino.plugin.jdbc.ForBaseJdbc;
 import io.trino.plugin.jdbc.JdbcClient;
+import io.trino.plugin.jdbc.VarcharPredicatePushdownModule;
 import io.trino.plugin.jdbc.credential.CredentialProvider;
 import org.mariadb.jdbc.Driver;
 
@@ -40,6 +41,7 @@ public void configure(Binder binder)
         binder.bind(JdbcClient.class).annotatedWith(ForBaseJdbc.class).to(MemSqlClient.class).in(Scopes.SINGLETON);
         configBinder(binder).bindConfig(MemSqlConfig.class);
         binder.install(new DecimalModule());
+        binder.install(new VarcharPredicatePushdownModule());
     }
 
     @Provides

plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClientModule.java

@@ -25,6 +25,7 @@
 import io.trino.plugin.jdbc.DriverConnectionFactory;
 import io.trino.plugin.jdbc.ForBaseJdbc;
 import io.trino.plugin.jdbc.JdbcClient;
+import io.trino.plugin.jdbc.VarcharPredicatePushdownModule;
 import io.trino.plugin.jdbc.credential.CredentialProvider;
 
 import java.sql.SQLException;
@@ -42,6 +43,7 @@ public void configure(Binder binder)
         configBinder(binder).bindConfig(MySqlJdbcConfig.class);
         configBinder(binder).bindConfig(MySqlConfig.class);
         binder.install(new DecimalModule());
+        binder.install(new VarcharPredicatePushdownModule());
     }
 
     @Provides

plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlClient.java

@@ -156,6 +156,7 @@
 import static io.trino.plugin.jdbc.TypeHandlingJdbcSessionProperties.getUnsupportedTypeHandling;
 import static io.trino.plugin.jdbc.UnsupportedTypeHandling.CONVERT_TO_VARCHAR;
 import static io.trino.plugin.jdbc.UnsupportedTypeHandling.IGNORE;
+import static io.trino.plugin.jdbc.VarcharPredicatePushdownSessionProperties.isUnsafeVarcharPushdownEnabled;
 import static io.trino.plugin.postgresql.PostgreSqlConfig.ArrayMapping.AS_ARRAY;
 import static io.trino.plugin.postgresql.PostgreSqlConfig.ArrayMapping.AS_JSON;
 import static io.trino.plugin.postgresql.PostgreSqlConfig.ArrayMapping.DISABLED;
@@ -228,6 +229,10 @@ public class PostgreSqlClient
                 domain.getType() instanceof VarcharType || domain.getType() instanceof CharType,
                 "This PredicatePushdownController can be used only for chars and varchars");
 
+        if (isUnsafeVarcharPushdownEnabled(session)) {
+            return FULL_PUSHDOWN.apply(session, domain);
+        }
+
         if (domain.isOnlyNull() ||
                 // PostgreSQL is case sensitive by default
                 domain.getValues().isDiscreteSet()) {

plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlClientModule.java

@@ -24,6 +24,7 @@
 import io.trino.plugin.jdbc.DriverConnectionFactory;
 import io.trino.plugin.jdbc.ForBaseJdbc;
 import io.trino.plugin.jdbc.JdbcClient;
+import io.trino.plugin.jdbc.VarcharPredicatePushdownModule;
 import io.trino.plugin.jdbc.credential.CredentialProvider;
 import org.postgresql.Driver;
 
@@ -40,6 +41,7 @@ public void configure(Binder binder)
         configBinder(binder).bindConfig(PostgreSqlConfig.class);
         bindSessionPropertiesProvider(binder, PostgreSqlSessionProperties.class);
         binder.install(new DecimalModule());
+        binder.install(new VarcharPredicatePushdownModule());
     }
 
     @Provides

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClientModule.java

@@ -26,6 +26,7 @@
 import io.trino.plugin.jdbc.ForBaseJdbc;
 import io.trino.plugin.jdbc.JdbcClient;
 import io.trino.plugin.jdbc.MaxDomainCompactionThreshold;
+import io.trino.plugin.jdbc.VarcharPredicatePushdownModule;
 import io.trino.plugin.jdbc.credential.CredentialProvider;
 
 import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
@@ -41,6 +42,7 @@ public void configure(Binder binder)
         binder.bind(JdbcClient.class).annotatedWith(ForBaseJdbc.class).to(SqlServerClient.class).in(Scopes.SINGLETON);
         bindTablePropertiesProvider(binder, SqlServerTableProperties.class);
         newOptionalBinder(binder, Key.get(int.class, MaxDomainCompactionThreshold.class)).setBinding().toInstance(SQL_SERVER_MAX_LIST_EXPRESSIONS);
+        binder.install(new VarcharPredicatePushdownModule());
     }
 
     @Provides