Add IN operator limit

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

@@ -40,6 +40,11 @@ public class JdbcMetadataConfig
     // between performance and pushdown capabilities
     private int domainCompactionThreshold = 32;
 
+    // The limit of the values per IN operator depends on the database. 
+    // E.g. Oracle allows only up to 1,000 IN list values in a SQL statement.
+    // A value of 0 means no limit
+    private int inOperatorLimit = 0;
+
     public boolean isAllowDropTable()
     {
         return allowDropTable;
@@ -107,4 +112,18 @@ public JdbcMetadataConfig setDomainCompactionThreshold(int domainCompactionThres
         this.domainCompactionThreshold = domainCompactionThreshold;
         return this;
     }
+
+    @Min(0)
+    public int getInOperatorLimit()
+    {
+        return inOperatorLimit;
+    }    
+
+    @Config("in-operator-limit")
+    @ConfigDescription("Maximum number of values per IN operator. A value of 0 means no limit")
+    public JdbcMetadataConfig setInOperatorLimit(int inOperatorLimit)
+    {
+        this.inOperatorLimit = inOperatorLimit;
+        return this;
+    }    
 }

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

@@ -36,6 +36,7 @@ public class JdbcMetadataSessionProperties
     public static final String AGGREGATION_PUSHDOWN_ENABLED = "aggregation_pushdown_enabled";
     public static final String TOPN_PUSHDOWN_ENABLED = "topn_pushdown_enabled";
     public static final String DOMAIN_COMPACTION_THRESHOLD = "domain_compaction_threshold";
+    public static final String IN_OPERATOR_LIMIT = "in_operator_limit";
 
     private final List<PropertyMetadata<?>> properties;
 
@@ -65,6 +66,12 @@ public JdbcMetadataSessionProperties(JdbcMetadataConfig jdbcMetadataConfig, @Max
                         "Enable TopN pushdown",
                         jdbcMetadataConfig.isTopNPushdownEnabled(),
                         false))
+                .add(integerProperty(
+                		IN_OPERATOR_LIMIT,
+                        "Maximum number of values per IN operator. A value of 0 means no limit",
+                        jdbcMetadataConfig.getInOperatorLimit(),
+                        value -> validateInOperatorLimit(value),
+                        false))
                 .build();
     }
 
@@ -94,6 +101,11 @@ public static int getDomainCompactionThreshold(ConnectorSession session)
         return session.getProperty(DOMAIN_COMPACTION_THRESHOLD, Integer.class);
     }
 
+    public static int getInOperatorLimit(ConnectorSession session)
+    {
+        return session.getProperty(IN_OPERATOR_LIMIT, Integer.class);
+    }
+
     private static void validateDomainCompactionThreshold(int domainCompactionThreshold, Optional<Integer> maxDomainCompactionThreshold)
     {
         if (domainCompactionThreshold < 1) {
@@ -106,4 +118,11 @@ private static void validateDomainCompactionThreshold(int domainCompactionThresh
             }
         });
     }
+
+    private static void validateInOperatorLimit(int inOperatorLimit)
+    {
+        if (inOperatorLimit < 0) {
+            throw new TrinoException(INVALID_SESSION_PROPERTY, format("%s must be greater than or equal to 0: %s", IN_OPERATOR_LIMIT, inOperatorLimit));
+        }
+    }
 }

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

@@ -40,6 +40,7 @@
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Verify.verify;
 import static com.google.common.collect.Iterables.getOnlyElement;
+import static io.trino.plugin.jdbc.JdbcMetadataSessionProperties.getInOperatorLimit;
 import static java.lang.String.format;
 import static java.util.Collections.nCopies;
 import static java.util.Objects.requireNonNull;
@@ -323,8 +324,20 @@ else if (singleValues.size() > 1) {
             for (Object value : singleValues) {
                 accumulator.accept(new QueryParameter(jdbcType, type, Optional.of(value)));
             }
-            String values = Joiner.on(",").join(nCopies(singleValues.size(), writeFunction.getBindExpression()));
-            disjuncts.add(client.quoted(column.getColumnName()) + " IN (" + values + ")");
+            if (getInOperatorLimit(session) == 0) {
+                String values = Joiner.on(",").join(nCopies(singleValues.size(), writeFunction.getBindExpression()));
+                disjuncts.add(client.quoted(column.getColumnName()) + " IN (" + values + ")");
+            }
+            else {
+                // The limit of the values per IN operator depends on the database. E.g. Oracle allows only up to 1,000 IN list values in a SQL statement. 
+            	int remainder = singleValues.size() % getInOperatorLimit(session);
+            	for (int i = 0; i < (singleValues.size() - remainder) / getInOperatorLimit(session); i++) {
+	                String values = Joiner.on(",").join(nCopies(getInOperatorLimit(session), writeFunction.getBindExpression()));
+	                disjuncts.add(client.quoted(column.getColumnName()) + " IN (" + values + ")");
+            	}
+                String values = Joiner.on(",").join(nCopies(remainder, writeFunction.getBindExpression()));
+                disjuncts.add(client.quoted(column.getColumnName()) + " IN (" + values + ")");
+            }
         }
 
         checkState(!disjuncts.isEmpty());