diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/expression/RewriteLikeEscapeWithCaseSensitivity.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/expression/RewriteLikeEscapeWithCaseSensitivity.java
new file mode 100644
index 000000000000..27f7264e6b8e
--- /dev/null
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/expression/RewriteLikeEscapeWithCaseSensitivity.java
@@ -0,0 +1,94 @@
+/*
+ * 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.expression;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.matching.Capture;
+import io.trino.matching.Captures;
+import io.trino.matching.Pattern;
+import io.trino.plugin.base.expression.ConnectorExpressionRule;
+import io.trino.plugin.jdbc.CaseSensitivity;
+import io.trino.plugin.jdbc.JdbcColumnHandle;
+import io.trino.plugin.jdbc.QueryParameter;
+import io.trino.spi.expression.Call;
+import io.trino.spi.expression.ConnectorExpression;
+import io.trino.spi.expression.Variable;
+import io.trino.spi.type.VarcharType;
+
+import java.util.Optional;
+
+import static io.trino.matching.Capture.newCapture;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.argument;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.argumentCount;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.call;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.expression;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.functionName;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.type;
+import static io.trino.plugin.jdbc.CaseSensitivity.CASE_INSENSITIVE;
+import static io.trino.spi.expression.StandardFunctions.LIKE_FUNCTION_NAME;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
+import static java.lang.String.format;
+
+public class RewriteLikeEscapeWithCaseSensitivity
+        implements ConnectorExpressionRule<Call, ParameterizedExpression>
+{
+    private static final Capture<ConnectorExpression> LIKE_VALUE = newCapture();
+    private static final Capture<ConnectorExpression> LIKE_PATTERN = newCapture();
+    private static final Capture<ConnectorExpression> ESCAPE_PATTERN = newCapture();
+    private static final Pattern<Call> PATTERN = call()
+            .with(functionName().equalTo(LIKE_FUNCTION_NAME))
+            .with(type().equalTo(BOOLEAN))
+            .with(argumentCount().equalTo(3))
+            .with(argument(0).matching(expression().capturedAs(LIKE_VALUE).with(type().matching(VarcharType.class::isInstance))))
+            .with(argument(1).matching(expression().capturedAs(LIKE_PATTERN).with(type().matching(VarcharType.class::isInstance))))
+            .with(argument(2).matching(expression().capturedAs(ESCAPE_PATTERN).with(type().matching(VarcharType.class::isInstance))));
+
+    @Override
+    public Pattern<Call> getPattern()
+    {
+        return PATTERN;
+    }
+
+    @Override
+    public Optional<ParameterizedExpression> rewrite(Call expression, Captures captures, RewriteContext<ParameterizedExpression> context)
+    {
+        ConnectorExpression capturedValue = captures.get(LIKE_VALUE);
+        if (capturedValue instanceof Variable variable) {
+            JdbcColumnHandle columnHandle = (JdbcColumnHandle) context.getAssignment(variable.getName());
+            Optional<CaseSensitivity> caseSensitivity = columnHandle.getJdbcTypeHandle().getCaseSensitivity();
+            if (caseSensitivity.orElse(CASE_INSENSITIVE) == CASE_INSENSITIVE) {
+                return Optional.empty();
+            }
+        }
+        Optional<ParameterizedExpression> value = context.defaultRewrite(capturedValue);
+        if (value.isEmpty()) {
+            return Optional.empty();
+        }
+
+        ImmutableList.Builder<QueryParameter> parameters = ImmutableList.builder();
+        parameters.addAll(value.get().parameters());
+        Optional<ParameterizedExpression> pattern = context.defaultRewrite(captures.get(LIKE_PATTERN));
+        if (pattern.isEmpty()) {
+            return Optional.empty();
+        }
+        parameters.addAll(pattern.get().parameters());
+
+        Optional<ParameterizedExpression> escape = context.defaultRewrite(captures.get(ESCAPE_PATTERN));
+        if (escape.isEmpty()) {
+            return Optional.empty();
+        }
+        parameters.addAll(escape.get().parameters());
+        return Optional.of(new ParameterizedExpression(format("%s LIKE %s ESCAPE %s", value.get().expression(), pattern.get().expression(), escape.get().expression()), parameters.build()));
+    }
+}
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/expression/RewriteLikeWithCaseSensitivity.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/expression/RewriteLikeWithCaseSensitivity.java
new file mode 100644
index 000000000000..5532af4eaa4b
--- /dev/null
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/expression/RewriteLikeWithCaseSensitivity.java
@@ -0,0 +1,86 @@
+/*
+ * 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.expression;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.matching.Capture;
+import io.trino.matching.Captures;
+import io.trino.matching.Pattern;
+import io.trino.plugin.base.expression.ConnectorExpressionRule;
+import io.trino.plugin.jdbc.CaseSensitivity;
+import io.trino.plugin.jdbc.JdbcColumnHandle;
+import io.trino.plugin.jdbc.QueryParameter;
+import io.trino.spi.expression.Call;
+import io.trino.spi.expression.ConnectorExpression;
+import io.trino.spi.expression.Variable;
+import io.trino.spi.type.VarcharType;
+
+import java.util.Optional;
+
+import static io.trino.matching.Capture.newCapture;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.argument;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.argumentCount;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.call;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.expression;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.functionName;
+import static io.trino.plugin.base.expression.ConnectorExpressionPatterns.type;
+import static io.trino.plugin.jdbc.CaseSensitivity.CASE_INSENSITIVE;
+import static io.trino.spi.expression.StandardFunctions.LIKE_FUNCTION_NAME;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
+import static java.lang.String.format;
+
+public class RewriteLikeWithCaseSensitivity
+        implements ConnectorExpressionRule<Call, ParameterizedExpression>
+{
+    private static final Capture<ConnectorExpression> LIKE_VALUE = newCapture();
+    private static final Capture<ConnectorExpression> LIKE_PATTERN = newCapture();
+    private static final Pattern<Call> PATTERN = call()
+            .with(functionName().equalTo(LIKE_FUNCTION_NAME))
+            .with(type().equalTo(BOOLEAN))
+            .with(argumentCount().equalTo(2))
+            .with(argument(0).matching(expression().capturedAs(LIKE_VALUE).with(type().matching(VarcharType.class::isInstance))))
+            .with(argument(1).matching(expression().capturedAs(LIKE_PATTERN).with(type().matching(VarcharType.class::isInstance))));
+
+    @Override
+    public Pattern<Call> getPattern()
+    {
+        return PATTERN;
+    }
+
+    @Override
+    public Optional<ParameterizedExpression> rewrite(Call expression, Captures captures, RewriteContext<ParameterizedExpression> context)
+    {
+        ConnectorExpression capturedValue = captures.get(LIKE_VALUE);
+        if (capturedValue instanceof Variable variable) {
+            JdbcColumnHandle columnHandle = (JdbcColumnHandle) context.getAssignment(variable.getName());
+            Optional<CaseSensitivity> caseSensitivity = columnHandle.getJdbcTypeHandle().getCaseSensitivity();
+            if (caseSensitivity.orElse(CASE_INSENSITIVE) == CASE_INSENSITIVE) {
+                return Optional.empty();
+            }
+        }
+        Optional<ParameterizedExpression> value = context.defaultRewrite(capturedValue);
+        if (value.isEmpty()) {
+            return Optional.empty();
+        }
+
+        ImmutableList.Builder<QueryParameter> parameters = ImmutableList.builder();
+        parameters.addAll(value.get().parameters());
+        Optional<ParameterizedExpression> pattern = context.defaultRewrite(captures.get(LIKE_PATTERN));
+        if (pattern.isEmpty()) {
+            return Optional.empty();
+        }
+        parameters.addAll(pattern.get().parameters());
+        return Optional.of(new ParameterizedExpression(format("%s LIKE %s", value.get().expression(), pattern.get().expression()), parameters.build()));
+    }
+}
diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/BaseTestRewriteLikeWithCaseSensitivity.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/BaseTestRewriteLikeWithCaseSensitivity.java
new file mode 100644
index 000000000000..fb7b8221193e
--- /dev/null
+++ b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/BaseTestRewriteLikeWithCaseSensitivity.java
@@ -0,0 +1,80 @@
+/*
+ * 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.expression;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.matching.Match;
+import io.trino.plugin.base.expression.ConnectorExpressionRule;
+import io.trino.plugin.jdbc.JdbcColumnHandle;
+import io.trino.plugin.jdbc.JdbcTypeHandle;
+import io.trino.plugin.jdbc.QueryParameter;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.expression.Call;
+import io.trino.spi.expression.ConnectorExpression;
+import io.trino.spi.expression.Variable;
+
+import java.sql.Types;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.google.common.collect.MoreCollectors.toOptional;
+import static io.trino.plugin.jdbc.CaseSensitivity.CASE_SENSITIVE;
+import static io.trino.plugin.jdbc.TestingJdbcTypeHandle.JDBC_BIGINT;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class BaseTestRewriteLikeWithCaseSensitivity
+{
+    protected abstract ConnectorExpressionRule<Call, ParameterizedExpression> getRewrite();
+
+    protected Optional<ParameterizedExpression> apply(Call expression)
+    {
+        Optional<Match> match = getRewrite().getPattern().match(expression).collect(toOptional());
+        if (match.isEmpty()) {
+            return Optional.empty();
+        }
+        return getRewrite().rewrite(expression, match.get().captures(), new ConnectorExpressionRule.RewriteContext<>()
+        {
+            @Override
+            public Map<String, ColumnHandle> getAssignments()
+            {
+                return Map.of("case_insensitive_value", new JdbcColumnHandle("case_insensitive_value", JDBC_BIGINT, VARCHAR),
+                        "case_sensitive_value", new JdbcColumnHandle("case_sensitive_value", new JdbcTypeHandle(Types.VARCHAR, Optional.of("varchar"), Optional.of(10), Optional.empty(), Optional.empty(), Optional.of(CASE_SENSITIVE)), VARCHAR));
+            }
+
+            @Override
+            public ConnectorSession getSession()
+            {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public Optional<ParameterizedExpression> defaultRewrite(ConnectorExpression expression)
+            {
+                if (expression instanceof Variable) {
+                    String name = ((Variable) expression).getName();
+                    return Optional.of(new ParameterizedExpression("\"" + name.replace("\"", "\"\"") + "\"", ImmutableList.of(new QueryParameter(expression.getType(), Optional.of(name)))));
+                }
+                return Optional.empty();
+            }
+        });
+    }
+
+    protected void assertNoRewrite(Call expression)
+    {
+        Optional<ParameterizedExpression> rewritten = apply(expression);
+        assertThat(rewritten).isEmpty();
+    }
+}
diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/TestRewriteLikeEscapeWithCaseSensitivity.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/TestRewriteLikeEscapeWithCaseSensitivity.java
new file mode 100644
index 000000000000..c564e2481514
--- /dev/null
+++ b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/TestRewriteLikeEscapeWithCaseSensitivity.java
@@ -0,0 +1,127 @@
+/*
+ * 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.expression;
+
+import io.trino.plugin.base.expression.ConnectorExpressionRule;
+import io.trino.plugin.jdbc.QueryParameter;
+import io.trino.spi.expression.Call;
+import io.trino.spi.expression.FunctionName;
+import io.trino.spi.expression.Variable;
+import org.testng.annotations.Test;
+
+import java.util.List;
+import java.util.Optional;
+
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class TestRewriteLikeEscapeWithCaseSensitivity
+        extends BaseTestRewriteLikeWithCaseSensitivity
+{
+    private final RewriteLikeEscapeWithCaseSensitivity rewrite = new RewriteLikeEscapeWithCaseSensitivity();
+
+    @Override
+    protected ConnectorExpressionRule<Call, ParameterizedExpression> getRewrite()
+    {
+        return rewrite;
+    }
+
+    @Test
+    public void testRewriteLikeEscapeCallInvalidNumberOfArguments()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(new Variable("case_sensitive_value", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeEscapeCallInvalidTypeValue()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_sensitive_value", BIGINT),
+                        new Variable("pattern", VARCHAR),
+                        new Variable("escape", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeEscapeCallInvalidTypePattern()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_sensitive_value", VARCHAR),
+                        new Variable("pattern", BIGINT),
+                        new Variable("escape", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeEscapeCallInvalidTypeEscape()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_sensitive_value", VARCHAR),
+                        new Variable("pattern", VARCHAR),
+                        new Variable("escape", BIGINT)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeEscapeCallOnCaseInsensitiveValue()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_insensitive_value", VARCHAR),
+                        new Variable("pattern", VARCHAR),
+                        new Variable("escape", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeEscapeCallOnCaseSensitiveValue()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_sensitive_value", VARCHAR),
+                        new Variable("pattern", VARCHAR),
+                        new Variable("escape", VARCHAR)));
+
+        ParameterizedExpression rewritten = apply(expression).orElseThrow();
+        assertThat(rewritten.expression()).isEqualTo("\"case_sensitive_value\" LIKE \"pattern\" ESCAPE \"escape\"");
+        assertThat(rewritten.parameters()).isEqualTo(List.of(
+                new QueryParameter(VARCHAR, Optional.of("case_sensitive_value")),
+                new QueryParameter(VARCHAR, Optional.of("pattern")),
+                new QueryParameter(VARCHAR, Optional.of("escape"))));
+    }
+}
diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/TestRewriteLikeWithCaseSensitivity.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/TestRewriteLikeWithCaseSensitivity.java
new file mode 100644
index 000000000000..d432479ca91e
--- /dev/null
+++ b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/expression/TestRewriteLikeWithCaseSensitivity.java
@@ -0,0 +1,119 @@
+/*
+ * 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.expression;
+
+import io.trino.plugin.base.expression.ConnectorExpressionRule;
+import io.trino.plugin.jdbc.QueryParameter;
+import io.trino.spi.expression.Call;
+import io.trino.spi.expression.FunctionName;
+import io.trino.spi.expression.Variable;
+import org.testng.annotations.Test;
+
+import java.util.List;
+import java.util.Optional;
+
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class TestRewriteLikeWithCaseSensitivity
+        extends BaseTestRewriteLikeWithCaseSensitivity
+{
+    private final RewriteLikeWithCaseSensitivity rewrite = new RewriteLikeWithCaseSensitivity();
+
+    @Override
+    protected ConnectorExpressionRule<Call, ParameterizedExpression> getRewrite()
+    {
+        return rewrite;
+    }
+
+    @Test
+    public void testRewriteLikeEscapeCallInvalidNumberOfArguments()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(new Variable("case_sensitive_value", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeCallInvalidNumberOfArguments()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(new Variable("case_sensitive_value", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeCallInvalidTypeValue()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_sensitive_value", BIGINT),
+                        new Variable("pattern", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeCallInvalidTypePattern()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_sensitive_value", VARCHAR),
+                        new Variable("pattern", BIGINT)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeCallOnCaseInsensitiveValue()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_insensitive_value", VARCHAR),
+                        new Variable("pattern", VARCHAR)));
+
+        assertNoRewrite(expression);
+    }
+
+    @Test
+    public void testRewriteLikeCallOnCaseSensitiveValue()
+    {
+        Call expression = new Call(
+                BOOLEAN,
+                new FunctionName("$like"),
+                List.of(
+                        new Variable("case_sensitive_value", VARCHAR),
+                        new Variable("pattern", VARCHAR)));
+
+        ParameterizedExpression rewritten = apply(expression).orElseThrow();
+        assertThat(rewritten.expression()).isEqualTo("\"case_sensitive_value\" LIKE \"pattern\"");
+        assertThat(rewritten.parameters()).isEqualTo(List.of(
+                new QueryParameter(VARCHAR, Optional.of("case_sensitive_value")),
+                new QueryParameter(VARCHAR, Optional.of("pattern"))));
+    }
+}
diff --git a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java
index 549ed23fc18b..4368b1342829 100644
--- a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java
+++ b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java
@@ -57,6 +57,8 @@
 import io.trino.plugin.jdbc.aggregation.ImplementVarianceSamp;
 import io.trino.plugin.jdbc.expression.JdbcConnectorExpressionRewriterBuilder;
 import io.trino.plugin.jdbc.expression.ParameterizedExpression;
+import io.trino.plugin.jdbc.expression.RewriteLikeEscapeWithCaseSensitivity;
+import io.trino.plugin.jdbc.expression.RewriteLikeWithCaseSensitivity;
 import io.trino.plugin.jdbc.logging.RemoteQueryModifier;
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.AggregateFunction;
@@ -68,6 +70,7 @@
 import io.trino.spi.connector.JoinType;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.connector.TableNotFoundException;
+import io.trino.spi.expression.ConnectorExpression;
 import io.trino.spi.predicate.Domain;
 import io.trino.spi.predicate.ValueSet;
 import io.trino.spi.statistics.ColumnStatistics;
@@ -250,6 +253,8 @@ public MySqlClient(
 
         this.connectorExpressionRewriter = JdbcConnectorExpressionRewriterBuilder.newBuilder()
                 .addStandardRules(this::quoted)
+                .add(new RewriteLikeWithCaseSensitivity())
+                .add(new RewriteLikeEscapeWithCaseSensitivity())
                 .build();
 
         JdbcTypeHandle bigintTypeHandle = new JdbcTypeHandle(Types.BIGINT, Optional.of("bigint"), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
@@ -270,6 +275,12 @@ public MySqlClient(
                         .build());
     }
 
+    @Override
+    public Optional<ParameterizedExpression> convertPredicate(ConnectorSession session, ConnectorExpression expression, Map<String, ColumnHandle> assignments)
+    {
+        return connectorExpressionRewriter.rewrite(session, expression, assignments);
+    }
+
     @Override
     protected Map<String, CaseSensitivity> getCaseSensitivityForColumns(ConnectorSession session, Connection connection, JdbcTableHandle tableHandle)
     {
diff --git a/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java b/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
index 0e7b8fb80489..8b2e7359cbb2 100644
--- a/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
+++ b/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
@@ -30,6 +30,7 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.time.LocalDate;
+import java.util.List;
 import java.util.Optional;
 import java.util.OptionalInt;
 
@@ -289,9 +290,53 @@ public void testAddNotNullColumn()
         }
     }
 
+    @Test
+    public void testLikePredicatePushdownWithCollation()
+    {
+        try (TestTable table = new TestTable(
+                onRemoteDatabase(),
+                "tpch.test_like_predicate_pushdown",
+                "(id integer, a_varchar varchar(1) CHARACTER SET utf8 COLLATE utf8_bin)",
+                List.of(
+                        "1, 'A'",
+                        "2, 'a'",
+                        "3, 'B'",
+                        "4, 'ą'",
+                        "5, 'Ą'"))) {
+            assertThat(query("SELECT id FROM " + table.getName() + " WHERE a_varchar LIKE '%A%'"))
+                    .isFullyPushedDown();
+            assertThat(query("SELECT id FROM " + table.getName() + " WHERE a_varchar LIKE '%ą%'"))
+                    .isFullyPushedDown();
+        }
+    }
+
+    @Test
+    public void testLikeWithEscapePredicatePushdownWithCollation()
+    {
+        try (TestTable table = new TestTable(
+                onRemoteDatabase(),
+                "tpch.test_like_with_escape_predicate_pushdown",
+                "(id integer, a_varchar varchar(4) CHARACTER SET utf8 COLLATE utf8_bin)",
+                List.of(
+                        "1, 'A%b'",
+                        "2, 'Asth'",
+                        "3, 'ą%b'",
+                        "4, 'ąsth'"))) {
+            assertThat(query("SELECT id FROM " + table.getName() + " WHERE a_varchar LIKE '%A\\%%' ESCAPE '\\'"))
+                    .isFullyPushedDown();
+            assertThat(query("SELECT id FROM " + table.getName() + " WHERE a_varchar LIKE '%ą\\%%' ESCAPE '\\'"))
+                    .isFullyPushedDown();
+        }
+    }
+
     @Test
     public void testPredicatePushdown()
     {
+        // varchar like
+        assertThat(query("SELECT regionkey, nationkey, name FROM nation WHERE name LIKE '%ROM%'"))
+                .matches("VALUES (BIGINT '3', BIGINT '19', CAST('ROMANIA' AS varchar(255)))")
+                .isNotFullyPushedDown(FilterNode.class);
+
         // varchar equality
         assertThat(query("SELECT regionkey, nationkey, name FROM nation WHERE name = 'ROMANIA'"))
                 .matches("VALUES (BIGINT '3', BIGINT '19', CAST('ROMANIA' AS varchar(255)))")
@@ -364,6 +409,10 @@ public void testPredicatePushdownWithCollation(String charset, String collation)
 
     private void testNationCollationQueries(String objectName)
     {
+        // varchar like
+        assertThat(query(format("SELECT regionkey, nationkey, name FROM %s WHERE name LIKE '%%ROM%%'", objectName)))
+                .isFullyPushedDown();
+
         // varchar inequality
         assertThat(query(format("SELECT regionkey, nationkey, name FROM %s WHERE name != 'ROMANIA' AND name != 'ALGERIA'", objectName)))
                 .isFullyPushedDown();