Add limit pushdown for Kafka connectors, and further refactor and enhance predicate pushdown in Kafka connector

[xxzhky]

Description

Add limit pushdown for Kafka connectors, and further refactor and enhance predicate pushdown in Kafka connector

Additional context and related issues

Add limit pushdown, and enhance predicate pushdown in Kafka connector

Significant changes have been made to improve predicate push-down support in Kafka connector. The code has been refactored to better handle intersection of old and new domains.

Additionally, the functionality to override partition end and begin offsets with range has been enhanced.

Lastly, an option to enable or disable predicate pushdown at runtime through session properties has been introduced.
By default, it supports predicate push-down. Push-down also can be disabled via kafka.predicate-force-push-down-enabled config prop or kafka.predicate_force_push_down_enabled session prop.

I try my test to make your life easier so that extensive comments have been added across multiple files in the Kafka connector for better readability and understanding of the code.

Release notes

(1) Add limit pushdown for Kafka connectors,
(2) Enhance predicate pushdown in Kafka connector.

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[xxzhky]

hello why has it gone？

…

---- Replied Message ---- | From | ***@***.***> | | Date | 01/11/2024 01:22 | | To | trinodb/trino ***@***.***> | | Cc | Fred ***@***.***>, Author ***@***.***> | | Subject | Re: [trinodb/trino] Add limit pushdown for Kafka connectors, and further refactor and enhance predicate pushdown in Kafka connector (PR #20146) | This pull request has gone a while without any activity. Tagging the Trino developer relations team: @***@***.***@mosabua — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you authored the thread.Message ID: ***@***.***>

[mosabua]

Could @wendigo @findepi @elonazoulay or anybody else with more knowledge about the Kafka connector help @xxzhky with a review please to move this PR forward.

Also @xxzhky can you confirm that the build passes locally and the CI failures are wrong, or otherwise fix these issues.

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[mosabua]

@xxzhky could you rebase and fix any build issues

[xxzhky]

@mosabua yes, I confirm that the build passes locally, however, some irrational checking issues happen to me. i.g.
ci / maven-checks

[wendigo]

I've rebased this PR locally and checked the dependency scope failure and here's the fix:

commit d896b0cf12d7b4c16818467166f225dbe38546fc
Author: Mateusz "Serafin" Gajewski <mateusz.gajewski@gmail.com>
Date:   Sun Feb 4 11:19:06 2024 +0100

    Fix

diff --git a/plugin/trino-kafka/pom.xml b/plugin/trino-kafka/pom.xml
index ce1fd85e77..6099a921f3 100644
--- a/plugin/trino-kafka/pom.xml
+++ b/plugin/trino-kafka/pom.xml
@@ -274,6 +274,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-parser</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-record-decoder</artifactId>
diff --git a/plugin/trino-kafka/src/test/java/io/trino/plugin/kafka/TestConstraintExtractor.java b/plugin/trino-kafka/src/test/java/io/trino/plugin/kafka/TestConstraintExtractor.java
index 3f98d51064..47e7a597c2 100644
--- a/plugin/trino-kafka/src/test/java/io/trino/plugin/kafka/TestConstraintExtractor.java
+++ b/plugin/trino-kafka/src/test/java/io/trino/plugin/kafka/TestConstraintExtractor.java
@@ -29,6 +29,7 @@ import io.trino.spi.type.TimestampType;
 import io.trino.spi.type.TimestampWithTimeZoneType;
 import io.trino.spi.type.Type;
 import io.trino.sql.planner.ConnectorExpressionTranslator;
+import io.trino.sql.planner.IrTypeAnalyzer;
 import io.trino.sql.planner.LiteralEncoder;
 import io.trino.sql.planner.Symbol;
 import io.trino.sql.planner.TypeProvider;
@@ -69,7 +70,6 @@ import static io.trino.spi.type.VarcharType.createVarcharType;
 import static io.trino.sql.analyzer.TypeSignatureProvider.fromTypes;
 import static io.trino.sql.analyzer.TypeSignatureTranslator.toSqlType;
 import static io.trino.sql.planner.TestingPlannerContext.PLANNER_CONTEXT;
-import static io.trino.sql.planner.TypeAnalyzer.createTestingTypeAnalyzer;
 import static io.trino.sql.tree.ComparisonExpression.Operator.EQUAL;
 import static io.trino.sql.tree.ComparisonExpression.Operator.GREATER_THAN;
 import static io.trino.sql.tree.ComparisonExpression.Operator.GREATER_THAN_OR_EQUAL;
@@ -743,7 +743,7 @@ public class TestConstraintExtractor
                         TypeProvider.viewOf(symbolTypes.entrySet().stream()
                                 .collect(toImmutableMap(entry -> new Symbol(entry.getKey()), Map.Entry::getValue))),
                         PLANNER_CONTEXT,
-                        createTestingTypeAnalyzer(PLANNER_CONTEXT))
+                        new IrTypeAnalyzer(PLANNER_CONTEXT))
                 .orElseThrow(() -> new RuntimeException("Translation to ConnectorExpression failed for: " + expression));
     }

[xxzhky]

@wendigo OK. Thanks very much, and could you like to help me fix it.

[xxzhky]

FYI，here attached the screenshot which the build passes locally.

[wendigo]

@xxzhky failure is in the dependency scope check, not tests. Tests are passing just fine on the CI.

[mosabua]

@xxzhky the diff in the comment above is the fix .. just apply that locally and amend the commit and force push.. then CI should pass

[xxzhky]

@mosabua @wendigo all checks done

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[xxzhky]

@mosabua @wendigo why has it gone again

[Praveen2112]

Is it possible to split them into smaller commits like

• Add limit pushdown
• Refactor
• Enhanced pushdown

This would help to ease up the review process

[xxzhky]

@Praveen2112 ok. I'll split them into more parts. Hong long could you get the review process done?

[Praveen2112]

I can start reviewing it by this week.

[Praveen2112]

It would be nice if we could have some test coverage in BaseConnectorTest

[Praveen2112]

Instead of converting ConnectorExpression into a TupleDomain - Can we ConnectorExpressionRewriter which would support the same for nested expression as well.

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[xxzhky]

@bitsondatadev @colebow @mosabua I am so busy recently, and I will handle it in the near future.

[mosabua]

Sounds good @xxzhky .. adding stale-ignore label.