support IN clause optimization for sharding key

features/sharding/api/src/main/java/org/apache/shardingsphere/sharding/spi/ShardingAlgorithm.java

@@ -18,7 +18,6 @@
 package org.apache.shardingsphere.sharding.spi;
 
 import org.apache.shardingsphere.infra.algorithm.core.ShardingSphereAlgorithm;
-
 import java.util.Optional;
 
 /**

features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/rewrite/token/ShardingTokenGenerateBuilder.java

@@ -33,6 +33,7 @@
 import org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingCursorTokenGenerator;
 import org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingDistinctProjectionPrefixTokenGenerator;
 import org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingFetchDirectionTokenGenerator;
+import org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingInPredicateTokenGenerator;
 import org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingIndexTokenGenerator;
 import org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingInsertValuesTokenGenerator;
 import org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingOffsetTokenGenerator;
@@ -78,6 +79,9 @@ public Collection<SQLTokenGenerator> getSQLTokenGenerators() {
         addSQLTokenGenerator(result, new ShardingRemoveTokenGenerator());
         addSQLTokenGenerator(result, new ShardingCursorTokenGenerator(rule));
         addSQLTokenGenerator(result, new ShardingFetchDirectionTokenGenerator());
+
+        // Add the IN predicate to optimize the Token generator
+        addSQLTokenGenerator(result, new ShardingInPredicateTokenGenerator(rule));
         return result;
     }
 

features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/rewrite/token/pojo/ShardingInPredicateToken.java

@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.shardingsphere.sharding.rewrite.token.pojo;
+
+import lombok.Getter;
+import org.apache.shardingsphere.infra.rewrite.sql.token.common.pojo.ParameterFilterable;
+import org.apache.shardingsphere.infra.rewrite.sql.token.common.pojo.RouteUnitAware;
+import org.apache.shardingsphere.infra.rewrite.sql.token.common.pojo.SQLToken;
+import org.apache.shardingsphere.infra.rewrite.sql.token.common.pojo.Substitutable;
+import org.apache.shardingsphere.infra.route.context.RouteUnit;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * SQL token for optimizing sharding IN predicates with value filtering per route unit.
+ * Simplified data structure with route information embedded in individual values.
+ */
+public final class ShardingInPredicateToken extends SQLToken implements Substitutable, RouteUnitAware, ParameterFilterable {
+
+    @Getter
+    private final int stopIndex;
+
+    /**
+     * Column name for this IN predicate.
+     */
+    @Getter
+    private final String columnName;
+
+    /**
+     * All values in this IN predicate with their route distribution information.
+     */
+    private final List<ShardingInPredicateValue> values;
+
+    public ShardingInPredicateToken(final int startIndex,
+                                    final int stopIndex,
+                                    final String columnName,
+                                    final List<ShardingInPredicateValue> values) {
+        super(startIndex);
+        this.stopIndex = stopIndex;
+        this.columnName = columnName;
+        this.values = values;
+    }
+
+    @Override
+    public String toString(final RouteUnit routeUnit) {
+        List<ShardingInPredicateValue> routeValues = getValuesForRoute(routeUnit);
+        if (routeValues.isEmpty()) {
+            return "";
+        }
+
+        return buildInClause(columnName, routeValues);
+    }
+
+    @Override
+    public Set<Integer> getRemovedParameterIndices(final RouteUnit routeUnit) {
+
+        // Add orphan parameter indices
+        Set<Integer> orphanIndices = values.stream()
+                .filter(ShardingInPredicateValue::isOrphan)
+                .filter(ShardingInPredicateValue::isParameter)
+                .map(ShardingInPredicateValue::getParameterIndex)
+                .collect(Collectors.toSet());
+        Set<Integer> result = new HashSet<>(orphanIndices);
+
+        // Add parameters that don't belong to this route
+        Set<Integer> routeParameterIndices = values.stream()
+                .filter(ShardingInPredicateValue::isParameter)
+                .filter(value -> value.belongsToRoute(routeUnit))
+                .map(ShardingInPredicateValue::getParameterIndex)
+                .collect(Collectors.toSet());
+
+        Set<Integer> nonRouteIndices = values.stream()
+                .filter(ShardingInPredicateValue::isParameter)
+                .map(ShardingInPredicateValue::getParameterIndex).collect(Collectors.toSet());
+        nonRouteIndices.removeAll(routeParameterIndices);
+        // Avoid double counting orphans
+        nonRouteIndices.removeAll(orphanIndices);
+        result.addAll(nonRouteIndices);
+
+        return result;
+    }
+
+    @Override
+    public boolean isParameterFilterable() {
+        return values.stream().anyMatch(ShardingInPredicateValue::isParameter);
+    }
+
+    /**
+     * Get values that belong to the specified route unit.
+     *
+     * @param routeUnit the route unit to filter values for
+     * @return a list of values that belong to the specified route unit
+     */
+    private List<ShardingInPredicateValue> getValuesForRoute(final RouteUnit routeUnit) {
+        return values.stream()
+                .filter(value -> value.belongsToRoute(routeUnit))
+                .collect(Collectors.toList());
+    }
+
+    /**
+     * Build optimized IN clause for the given values.
+     *
+     * @param column the column name for the IN clause
+     * @param valueList the list of values to include in the IN clause
+     * @return the optimized IN clause as a string
+     */
+    private String buildInClause(final String column, final List<ShardingInPredicateValue> valueList) {
+        if (valueList.size() == 1) {
+            ShardingInPredicateValue single = valueList.get(0);
+            return String.format("%s = %s", column,
+                    single.isParameter() ? "?" : formatValue(single.getValue()));
+        }
+        String valueString = valueList.stream()
+                .map(value -> value.isParameter() ? "?" : formatValue(value.getValue()))
+                .collect(Collectors.joining(", "));
+        return String.format("%s IN (%s)", column, valueString);
+    }
+
+    /**
+     * Format a value for SQL output.
+     *
+     * @param value the value to format
+     * @return the formatted value as a string
+     */
+    private String formatValue(final Comparable<?> value) {
+        if (null == value) {
+            return "NULL";
+        }
+        if (value instanceof String) {
+            return "'" + value.toString().replace("'", "''") + "'";
+        }
+        return value.toString();
+    }
+}

features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/rewrite/token/pojo/ShardingInPredicateValue.java

@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.shardingsphere.sharding.rewrite.token.pojo;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.infra.route.context.RouteUnit;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * Represents a value within a sharding IN predicate with its route distribution information.
+ * This class encapsulates not only the value information but also which route units this value belongs to.
+ *
+ * <p>In ShardingSphere's SQL rewriting process, IN predicates need special handling for sharding optimization.
+ * Each value in an IN clause can be either:
+ * <ul>
+ *   <li>A parameter marker (?): represented with parameterIndex and isParameter=true</li>
+ *   <li>A literal value: represented with the actual value and isParameter=false</li>
+ * </ul>
+ * Additionally, each value now knows which route units it belongs to, simplifying the overall data structure.
+ * </p>
+ *
+ * @author yinh
+ * @see org.apache.shardingsphere.sharding.rewrite.token.pojo.ShardingInPredicateToken
+ * @see org.apache.shardingsphere.sharding.rewrite.token.generator.impl.ShardingInPredicateTokenGenerator
+ */
+@RequiredArgsConstructor
+@Getter
+public final class ShardingInPredicateValue {
+
+    /**
+     * The index of the parameter marker in the original prepared statement.
+     * Only meaningful when {@link #isParameter} is true.
+     */
+    private final int parameterIndex;
+
+    /**
+     * The actual value of this predicate component.
+     * For parameter markers, this represents the bound parameter value.
+     * For literals, this represents the literal value from the SQL.
+     * Must implement Comparable for sharding algorithm processing.
+     */
+    private final Comparable<?> value;
+
+    /**
+     * Indicates whether this value originates from a parameter marker (?) in the SQL.
+     * When true, this value should be represented as "?" in rewritten SQL.
+     * When false, this value should be formatted as a literal in rewritten SQL.
+     */
+    private final boolean isParameter;
+
+    /**
+     * Set of route units that this value belongs to.
+     * During SQL rewriting, this value will only be included in the IN clauses
+     * for the route units specified in this set.
+     */
+    private final Set<RouteUnit> targetRoutes;
+
+    /**
+     * Indicates whether this value is an orphan parameter that doesn't belong to any route unit.
+     * Orphan parameters are those that cannot be mapped to any specific shard based on sharding algorithm.
+     */
+    private final boolean isOrphan;
+
+    /**
+     * Convenience constructor for non-orphan values.
+     */
+    public ShardingInPredicateValue(final int parameterIndex, final Comparable<?> value,
+                                    final boolean isParameter, final Set<RouteUnit> targetRoutes) {
+        this(parameterIndex, value, isParameter, targetRoutes, false);
+    }
+
+    /**
+     * Convenience constructor for orphan parameters.
+     *
+     * @param parameterIndex the index of the parameter marker in the original prepared statement
+     * @param value the actual value of this predicate component
+     * @param isParameter true if this value originates from a parameter marker, false if it's a literal
+     * @return a new ShardingInPredicateValue instance representing an orphan parameter
+     */
+    public static ShardingInPredicateValue createOrphan(final int parameterIndex, final Comparable<?> value, final boolean isParameter) {
+        return new ShardingInPredicateValue(parameterIndex, value, isParameter, Collections.emptySet(), true);
+    }
+
+    /**
+     * Check if this value belongs to the specified route unit.
+     *
+     * @param routeUnit the route unit to check
+     * @return true if this value belongs to the specified route unit
+     */
+    public boolean belongsToRoute(final RouteUnit routeUnit) {
+        return targetRoutes.contains(routeUnit);
+    }
+}

features/sharding/core/src/test/java/org/apache/shardingsphere/sharding/rewrite/token/ShardingTokenGenerateBuilderTest.java

@@ -32,6 +32,7 @@
 import org.mockito.internal.configuration.plugins.Plugins;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 
 import static org.hamcrest.CoreMatchers.is;
@@ -57,6 +58,7 @@ void setup() {
     void assertGetSQLTokenGenerators() throws Exception {
         when(routeContext.containsTableSharding()).thenReturn(true);
         SelectStatementContext sqlStatementContext = mock(SelectStatementContext.class, RETURNS_DEEP_STUBS);
+        when(sqlStatementContext.getWhereSegments()).thenReturn(Collections.emptyList());
         when(sqlStatementContext.getProjectionsContext().getAggregationProjections().isEmpty()).thenReturn(false);
         when(sqlStatementContext.getSqlStatement().getAttributes()).thenReturn(new SQLStatementAttributes());
         ShardingTokenGenerateBuilder shardingTokenGenerateBuilder = new ShardingTokenGenerateBuilder(shardingRule, routeContext, sqlStatementContext);