Repository: apache/amoro (java)

Issue:
[Feature]: Add the self-optimizing.partition-filter parameter.
### Description

Specify the partition range for Amoro optimization by adding self-optimizing.partition-filter, similar to the function of the where parameter in Spark's rewrite_data_files procedures.

### Use case/motivation

Currently, when Amoro optimizes Iceberg tables, it defaults to optimizing data from all partitions. However, this can bring about the following issues in practical use:

High cost of optimizing historical data: Some tables' historical data may not conform to Amoro's optimization rules, and optimizing all historical data can lead to resource waste and performance degradation.

Conflict between concurrent writing and optimization: Some historical partitions may have data repair operations involving deletions, which can take a relatively long duration. In such cases, it's preferable for Amoro to skip the optimization of these partitions to avoid conflicts.

### Describe the solution

Add the self-optimizing.partition-filter parameter.

### Subtasks

- [ ] Add the self-optimizing.partition-filter parameter. @lintingbin 

### Related issues

_No response_

### Are you willing to submit a PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)


Pull request description:
[AMORO-3424] Add the self-optimizing.partition-filter parameter
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://amoro.apache.org/how-to-contribute/
  2. If the PR is related to an issue in https://github.com/apache/amoro/issues, add '[AMORO-XXXX]' in your PR title, e.g., '[AMORO-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][AMORO-XXXX] Your PR title ...'.
-->

## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about its use case.
  2. If you fix a bug, you can clarify why it is a bug.
  3. Use Fix/Resolve/Close #{ISSUE_NUMBER} to link this PR to its related issue
-->

Close #3424 .

## Brief change log
<!--
Clearly describe the changes made in modules, classes, methods, etc.
-->

- Add the self-optimizing.partition-filter parameter.

## How was this patch tested?

- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] Run test locally before making a pull request

## Documentation

- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs


Patch:
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java
index bb7823c99..157674922 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java
@@ -294,6 +294,11 @@ public class TableConfigurations {
                 properties,
                 TableProperties.SELF_OPTIMIZING_FULL_REWRITE_ALL_FILES,
                 TableProperties.SELF_OPTIMIZING_FULL_REWRITE_ALL_FILES_DEFAULT))
+        .setFilter(
+            CompatiblePropertyUtil.propertyAsString(
+                properties,
+                TableProperties.SELF_OPTIMIZING_FILTER,
+                TableProperties.SELF_OPTIMIZING_FILTER_DEFAULT))
         .setBaseHashBucket(
             CompatiblePropertyUtil.propertyAsInt(
                 properties,
diff --git a/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java b/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java
index ef92ac5fb..591e551c3 100644
--- a/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java
+++ b/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java
@@ -74,6 +74,9 @@ public class OptimizingConfig {
   // self-optimizing.full.rewrite-all-files
   private boolean fullRewriteAllFiles;
 
+  // self-optimizing.filter
+  private String filter;
+
   // base.file-index.hash-bucket
   private int baseHashBucket;
 
@@ -240,6 +243,15 @@ public class OptimizingConfig {
     return this;
   }
 
+  public OptimizingConfig setFilter(String filter) {
+    this.filter = filter;
+    return this;
+  }
+
+  public String getFilter() {
+    return filter;
+  }
+
   public int getBaseHashBucket() {
     return baseHashBucket;
   }
@@ -291,6 +303,7 @@ public class OptimizingConfig {
         && Double.compare(that.majorDuplicateRatio, majorDuplicateRatio) == 0
         && fullTriggerInterval == that.fullTriggerInterval
         && fullRewriteAllFiles == that.fullRewriteAllFiles
+        && Objects.equal(filter, that.filter)
         && baseHashBucket == that.baseHashBucket
         && baseRefreshInterval == that.baseRefreshInterval
         && hiveRefreshInterval == that.hiveRefreshInterval
@@ -317,6 +330,7 @@ public class OptimizingConfig {
         majorDuplicateRatio,
         fullTriggerInterval,
         fullRewriteAllFiles,
+        filter,
         baseHashBucket,
         baseRefreshInterval,
         hiveRefreshInterval,
@@ -341,6 +355,7 @@ public class OptimizingConfig {
         .add("majorDuplicateRatio", majorDuplicateRatio)
         .add("fullTriggerInterval", fullTriggerInterval)
         .add("fullRewriteAllFiles", fullRewriteAllFiles)
+        .add("filter", filter)
         .add("baseHashBucket", baseHashBucket)
         .add("baseRefreshInterval", baseRefreshInterval)
         .add("hiveRefreshInterval", hiveRefreshInterval)
diff --git a/amoro-format-iceberg/pom.xml b/amoro-format-iceberg/pom.xml
index 47919fce4..c76a879dd 100644
--- a/amoro-format-iceberg/pom.xml
+++ b/amoro-format-iceberg/pom.xml
@@ -154,6 +154,11 @@
             <artifactId>paimon-bundle</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.github.jsqlparser</groupId>
+            <artifactId>jsqlparser</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.roaringbitmap</groupId>
             <artifactId>RoaringBitmap</artifactId>
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingEvaluator.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingEvaluator.java
index abd30b002..6dfdf89ea 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingEvaluator.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingEvaluator.java
@@ -32,12 +32,12 @@ import org.apache.amoro.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgno
 import org.apache.amoro.table.KeyedTableSnapshot;
 import org.apache.amoro.table.MixedTable;
 import org.apache.amoro.table.TableSnapshot;
+import org.apache.amoro.utils.ExpressionUtil;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.SnapshotSummary;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.PropertyUtil;
@@ -111,7 +111,8 @@ public abstract class AbstractOptimizingEvaluator {
   }
 
   protected Expression getPartitionFilter() {
-    return Expressions.alwaysTrue();
+    return ExpressionUtil.convertSqlFilterToIcebergExpression(
+        config.getFilter(), mixedTable.schema().columns());
   }
 
   private void initPartitionPlans(TableFileScanHelper tableFileScanHelper) {
@@ -142,7 +143,7 @@ public abstract class AbstractOptimizingEvaluator {
         partitionPlanMap.entrySet().stream()
             .filter(entry -> entry.getValue().isNecessary())
             .limit(maxPendingPartitions)
-            .collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue())));
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
   }
 
   protected abstract PartitionEvaluator buildEvaluator(Pair<Integer, StructLike> partition);
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java
index f408f0c4f..0e4b1b0db 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java
@@ -31,6 +31,7 @@ import org.apache.amoro.utils.MixedTableUtil;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.util.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -110,6 +111,10 @@ public abstract class AbstractOptimizingPlanner extends AbstractOptimizingEvalua
 
   @Override
   protected Expression getPartitionFilter() {
+    if (Expressions.alwaysTrue().equals(partitionFilter)
+        && !Expressions.alwaysTrue().equals(super.getPartitionFilter())) {
+      return super.getPartitionFilter();
+    }
     return partitionFilter;
   }
 
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java
index 9f2aeca5d..08f2ed242 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java
@@ -115,6 +115,9 @@ public class TableProperties {
       "self-optimizing.full.rewrite-all-files";
   public static final boolean SELF_OPTIMIZING_FULL_REWRITE_ALL_FILES_DEFAULT = true;
 
+  public static final String SELF_OPTIMIZING_FILTER = "self-optimizing.filter";
+  public static final String SELF_OPTIMIZING_FILTER_DEFAULT = null;
+
   public static final String SELF_OPTIMIZING_MIN_PLAN_INTERVAL =
       "self-optimizing.min-plan-interval";
   public static final long SELF_OPTIMIZING_MIN_PLAN_INTERVAL_DEFAULT = 60000;
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/ExpressionUtil.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/ExpressionUtil.java
index 025f9e670..370a45532 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/ExpressionUtil.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/ExpressionUtil.java
@@ -18,7 +18,28 @@
 
 package org.apache.amoro.utils;
 
+import net.sf.jsqlparser.expression.DateTimeLiteralExpression;
+import net.sf.jsqlparser.expression.DoubleValue;
+import net.sf.jsqlparser.expression.LongValue;
+import net.sf.jsqlparser.expression.NotExpression;
+import net.sf.jsqlparser.expression.StringValue;
+import net.sf.jsqlparser.expression.operators.conditional.AndExpression;
+import net.sf.jsqlparser.expression.operators.conditional.OrExpression;
+import net.sf.jsqlparser.expression.operators.relational.EqualsTo;
+import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
+import net.sf.jsqlparser.expression.operators.relational.GreaterThan;
+import net.sf.jsqlparser.expression.operators.relational.GreaterThanEquals;
+import net.sf.jsqlparser.expression.operators.relational.InExpression;
+import net.sf.jsqlparser.expression.operators.relational.IsNullExpression;
+import net.sf.jsqlparser.expression.operators.relational.MinorThan;
+import net.sf.jsqlparser.expression.operators.relational.MinorThanEquals;
+import net.sf.jsqlparser.expression.operators.relational.NotEqualsTo;
+import net.sf.jsqlparser.parser.CCJSqlParserUtil;
+import net.sf.jsqlparser.schema.Column;
+import net.sf.jsqlparser.statement.select.PlainSelect;
+import net.sf.jsqlparser.statement.select.Select;
 import org.apache.amoro.table.MixedTable;
+import org.apache.commons.lang.StringUtils;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -28,7 +49,16 @@ import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.UnboundTerm;
 import org.apache.iceberg.types.Types;
 
+import java.sql.Date;
+import java.sql.Time;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoField;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
 
 /** Utility class for working with {@link Expression}. */
 public class ExpressionUtil {
@@ -79,4 +109,169 @@ public class ExpressionUtil {
     }
     return filter;
   }
+
+  /**
+   * Convert SQL filter to Iceberg {@link Expression}.
+   *
+   * @param sqlFilter the SQL filter
+   * @param tableColumns the list of table columns
+   * @return the Iceberg {@link Expression}
+   */
+  public static Expression convertSqlFilterToIcebergExpression(
+      String sqlFilter, List<Types.NestedField> tableColumns) {
+    if (StringUtils.isEmpty(StringUtils.trim(sqlFilter))) {
+      return Expressions.alwaysTrue();
+    }
+
+    try {
+      Select statement = (Select) CCJSqlParserUtil.parse("SELECT * FROM dummy WHERE " + sqlFilter);
+      PlainSelect select = statement.getPlainSelect();
+      return convertSparkExpressionToIceberg(select.getWhere(), tableColumns);
+    } catch (Exception e) {
+      throw new IllegalArgumentException("Failed to parse where condition: " + sqlFilter, e);
+    }
+  }
+
+  private static Expression convertSparkExpressionToIceberg(
+      net.sf.jsqlparser.expression.Expression whereExpr, List<Types.NestedField> tableColumns) {
+    if (whereExpr instanceof IsNullExpression) {
+      IsNullExpression isNull = (IsNullExpression) whereExpr;
+      Types.NestedField column = getColumn(isNull.getLeftExpression(), tableColumns);
+      return isNull.isNot()
+          ? Expressions.notNull(column.name())
+          : Expressions.isNull(column.name());
+    } else if (whereExpr instanceof EqualsTo) {
+      EqualsTo eq = (EqualsTo) whereExpr;
+      Types.NestedField column = getColumn(eq.getLeftExpression(), tableColumns);
+      return Expressions.equal(column.name(), getValue(eq.getRightExpression(), column));
+    } else if (whereExpr instanceof NotEqualsTo) {
+      NotEqualsTo ne = (NotEqualsTo) whereExpr;
+      Types.NestedField column = getColumn(ne.getLeftExpression(), tableColumns);
+      return Expressions.notEqual(column.name(), getValue(ne.getRightExpression(), column));
+    } else if (whereExpr instanceof GreaterThan) {
+      GreaterThan gt = (GreaterThan) whereExpr;
+      Types.NestedField column = getColumn(gt.getLeftExpression(), tableColumns);
+      return Expressions.greaterThan(column.name(), getValue(gt.getRightExpression(), column));
+    } else if (whereExpr instanceof GreaterThanEquals) {
+      GreaterThanEquals ge = (GreaterThanEquals) whereExpr;
+      Types.NestedField column = getColumn(ge.getLeftExpression(), tableColumns);
+      return Expressions.greaterThanOrEqual(
+          column.name(), getValue(ge.getRightExpression(), column));
+    } else if (whereExpr instanceof MinorThan) {
+      MinorThan lt = (MinorThan) whereExpr;
+      Types.NestedField column = getColumn(lt.getLeftExpression(), tableColumns);
+      return Expressions.lessThan(column.name(), getValue(lt.getRightExpression(), column));
+    } else if (whereExpr instanceof MinorThanEquals) {
+      MinorThanEquals le = (MinorThanEquals) whereExpr;
+      Types.NestedField column = getColumn(le.getLeftExpression(), tableColumns);
+      return Expressions.lessThanOrEqual(column.name(), getValue(le.getRightExpression(), column));
+    } else if (whereExpr instanceof InExpression) {
+      InExpression in = (InExpression) whereExpr;
+      Types.NestedField column = getColumn(in.getLeftExpression(), tableColumns);
+      net.sf.jsqlparser.expression.Expression rightExpr = in.getRightExpression();
+      List<Object> values = new ArrayList<>();
+      if (rightExpr instanceof ExpressionList) {
+        for (net.sf.jsqlparser.expression.Expression expr : ((ExpressionList<?>) rightExpr)) {
+          values.add(getValue(expr, column));
+        }
+      } else {
+        throw new UnsupportedOperationException("Subquery IN not supported");
+      }
+      return in.isNot()
+          ? Expressions.notIn(column.name(), values)
+          : Expressions.in(column.name(), values);
+    } else if (whereExpr instanceof NotExpression) {
+      NotExpression not = (NotExpression) whereExpr;
+      return Expressions.not(convertSparkExpressionToIceberg(not.getExpression(), tableColumns));
+    } else if (whereExpr instanceof AndExpression) {
+      AndExpression and = (AndExpression) whereExpr;
+      return Expressions.and(
+          convertSparkExpressionToIceberg(and.getLeftExpression(), tableColumns),
+          convertSparkExpressionToIceberg(and.getRightExpression(), tableColumns));
+    } else if (whereExpr instanceof OrExpression) {
+      OrExpression or = (OrExpression) whereExpr;
+      return Expressions.or(
+          convertSparkExpressionToIceberg(or.getLeftExpression(), tableColumns),
+          convertSparkExpressionToIceberg(or.getRightExpression(), tableColumns));
+    }
+    throw new UnsupportedOperationException("Unsupported expression: " + whereExpr);
+  }
+
+  private static Types.NestedField getColumn(
+      net.sf.jsqlparser.expression.Expression expr, List<Types.NestedField> tableColumns) {
+    if (expr instanceof Column) {
+      String columnName = ((Column) expr).getColumnName();
+      Optional<Types.NestedField> column =
+          tableColumns.stream().filter(c -> c.name().equals(columnName)).findFirst();
+      if (column.isPresent()) {
+        return column.get();
+      }
+      throw new IllegalArgumentException("Column not found: " + columnName);
+    }
+
+    throw new IllegalArgumentException("Expected column reference, got: " + expr);
+  }
+
+  private static Object getValue(
+      net.sf.jsqlparser.expression.Expression expr, Types.NestedField column) {
+    try {
+      return convertValue(expr, column);
+    } catch (Exception e) {
+      throw new IllegalArgumentException("Failed to convert value: " + expr, e);
+    }
+  }
+
+  private static Object convertValue(
+      net.sf.jsqlparser.expression.Expression expr, Types.NestedField column) {
+    switch (column.type().typeId()) {
+      case BOOLEAN:
+        return Boolean.valueOf(((Column) expr).getColumnName());
+      case STRING:
+        return ((StringValue) expr).getValue();
+      case INTEGER:
+      case LONG:
+        return ((LongValue) expr).getValue();
+      case FLOAT:
+      case DOUBLE:
+        return ((DoubleValue) expr).getValue();
+      case DATE:
+        String dateStr = getDateTimeLiteralStr(expr, "date");
+        if (dateStr != null) {
+          return Date.valueOf(dateStr).toLocalDate().toEpochDay();
+        }
+        break;
+      case TIME:
+        String timeStr = getDateTimeLiteralStr(expr, "time");
+        if (timeStr != null) {
+          return Time.valueOf(timeStr).toLocalTime().getLong(ChronoField.MICRO_OF_DAY);
+        }
+        break;
+      case TIMESTAMP:
+        String timestampStr = getDateTimeLiteralStr(expr, "timestamp");
+        if (timestampStr != null) {
+          long timestamp;
+          if (column.type().equals(Types.TimestampType.withZone())) {
+            timestamp = OffsetDateTime.parse(timestampStr).toEpochSecond();
+          } else {
+            timestamp = LocalDateTime.parse(timestampStr).toEpochSecond(ZoneOffset.ofHours(0));
+          }
+          return timestamp * 1_000_000L;
+        }
+        break;
+    }
+    throw new IllegalArgumentException(
+        expr + " can not be converted to column type: " + column.type());
+  }
+
+  private static String getDateTimeLiteralStr(
+      net.sf.jsqlparser.expression.Expression expr, String type) {
+    String timestampStr = null;
+    if (expr instanceof StringValue) {
+      timestampStr = ((StringValue) expr).getValue();
+    } else if (expr instanceof DateTimeLiteralExpression
+        && ((DateTimeLiteralExpression) expr).getType().name().equalsIgnoreCase(type)) {
+      timestampStr = ((DateTimeLiteralExpression) expr).getValue().replaceAll("^'(.*)'$", "$1");
+    }
+    return timestampStr;
+  }
 }
diff --git a/amoro-web/src/hooks/usePagination.ts b/amoro-web/src/hooks/usePagination.ts
index c5337224f..79b42467f 100644
--- a/amoro-web/src/hooks/usePagination.ts
+++ b/amoro-web/src/hooks/usePagination.ts
@@ -15,12 +15,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+import { useI18n } from 'vue-i18n'
 
 export function usePagination() {
   const _total = 0
   const _current = 1
   const _pageSizeOptions = ['25', '50', '100']
   const _pageSize = 25
+  const { t } = useI18n()
+
   return {
     total: _total,
     current: _current,
@@ -29,6 +32,6 @@ export function usePagination() {
     showQuickJumper: true,
     showSizeChanger: true,
     hideOnSinglePage: false,
-    // showTotal: (total: number) => `total ${total} records`
+    showTotal: (total: number) => { return t('totalCount', { total }) }
   }
 }
diff --git a/amoro-web/src/language/en.ts b/amoro-web/src/language/en.ts
index efea76e99..90901671d 100644
--- a/amoro-web/src/language/en.ts
+++ b/amoro-web/src/language/en.ts
@@ -225,4 +225,5 @@ export default {
   Failed: 'Failed',
   Finished: 'Finished',
   Created: 'Created',
+  totalCount: 'total {total} records',
 }
diff --git a/amoro-web/src/language/zh.ts b/amoro-web/src/language/zh.ts
index e59ff40f8..5d8d78b38 100644
--- a/amoro-web/src/language/zh.ts
+++ b/amoro-web/src/language/zh.ts
@@ -225,4 +225,5 @@ export default {
   Failed: '已失败',
   Finished: '已结束',
   Created: '已创建',
+  totalCount: '共 {total} 条',
 }
diff --git a/docs/user-guides/configurations.md b/docs/user-guides/configurations.md
index 5bcf7afbb..5fcde8842 100644
--- a/docs/user-guides/configurations.md
+++ b/docs/user-guides/configurations.md
@@ -43,23 +43,24 @@ modified through [Alter Table](../using-tables/#modify-table) operations.
 
 Self-optimizing configurations are applicable to both Iceberg Format and Mixed streaming Format.
 
-| Key                                           | Default          | Description                                                                                                                              |
-|-----------------------------------------------|------------------|------------------------------------------------------------------------------------------------------------------------------------------|
-| self-optimizing.enabled                       | true             | Enables Self-optimizing                                                                                                                  |
-| self-optimizing.group                         | default          | Optimizer group for Self-optimizing                                                                                                      |
-| self-optimizing.quota                         | 0.1              | Quota for Self-optimizing, indicating the CPU resource the table can take up                                                             |
-| self-optimizing.execute.num-retries           | 5                | Number of retries after failure of Self-optimizing                                                                                       |
-| self-optimizing.target-size                   | 134217728(128MB) | Target size for Self-optimizing                                                                                                          |
-| self-optimizing.max-file-count                | 10000            | Maximum number of files processed by a Self-optimizing process                                                                           |
-| self-optimizing.max-task-size-bytes           | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks                                                                             |
-| self-optimizing.fragment-ratio                | 8                | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size         |
-| self-optimizing.min-target-size-ratio         | 0.75             | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting                          |
-| self-optimizing.minor.trigger.file-count      | 12               | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count   |
-| self-optimizing.minor.trigger.interval        | 3600000(1 hour)  | The time interval in milliseconds to trigger minor optimizing                                                                            |
-| self-optimizing.major.trigger.duplicate-ratio | 0.1              | The ratio of duplicate data of segment files to trigger major optimizing                                                                 |
-| self-optimizing.full.trigger.interval         | -1(closed)       | The time interval in milliseconds to trigger full optimizing                                                                             |
-| self-optimizing.full.rewrite-all-files        | true             | Whether full optimizing rewrites all files or skips files that do not need to be optimized                                               |
-| self-optimizing.min-plan-interval             | 60000            | The minimum time interval between two self-optimizing planning action                                                                    |
+| Key                                           | Default          | Description                                                                                                                                                                                                                             |
+|-----------------------------------------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| self-optimizing.enabled                       | true             | Enables Self-optimizing                                                                                                                                                                                                                 |
+| self-optimizing.group                         | default          | Optimizer group for Self-optimizing                                                                                                                                                                                                     |
+| self-optimizing.quota                         | 0.1              | Quota for Self-optimizing, indicating the CPU resource the table can take up                                                                                                                                                            |
+| self-optimizing.execute.num-retries           | 5                | Number of retries after failure of Self-optimizing                                                                                                                                                                                      |
+| self-optimizing.target-size                   | 134217728(128MB) | Target size for Self-optimizing                                                                                                                                                                                                         |
+| self-optimizing.max-file-count                | 10000            | Maximum number of files processed by a Self-optimizing process                                                                                                                                                                          |
+| self-optimizing.max-task-size-bytes           | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks                                                                                                                                                                            |
+| self-optimizing.fragment-ratio                | 8                | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size                                                                                                        |
+| self-optimizing.min-target-size-ratio         | 0.75             | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting                                                                                                                         |
+| self-optimizing.minor.trigger.file-count      | 12               | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count                                                                                                  |
+| self-optimizing.minor.trigger.interval        | 3600000(1 hour)  | The time interval in milliseconds to trigger minor optimizing                                                                                                                                                                           |
+| self-optimizing.major.trigger.duplicate-ratio | 0.1              | The ratio of duplicate data of segment files to trigger major optimizing                                                                                                                                                                |
+| self-optimizing.full.trigger.interval         | -1(closed)       | The time interval in milliseconds to trigger full optimizing                                                                                                                                                                            |
+| self-optimizing.full.rewrite-all-files        | true             | Whether full optimizing rewrites all files or skips files that do not need to be optimized                                                                                                                                              |
+| self-optimizing.min-plan-interval             | 60000            | The minimum time interval between two self-optimizing planning action                                                                                                                                                                   |
+| self-optimizing.filter                        | NULL             | Filter conditions for self-optimizing, using SQL conditional expressions, without supporting any functions. For the timestamp column condition, the ISO date-time formatter must be used. For example: op_time > '2007-12-03T10:15:30'. |
 
 ## Data-cleaning configurations
 
diff --git a/pom.xml b/pom.xml
index 458481f38..e7e1533b4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -151,6 +151,7 @@
         <guava.version>32.1.1-jre</guava.version>
         <hudi.version>0.14.1</hudi.version>
         <pagehelper.version>6.1.0</pagehelper.version>
+        <jsqlparser.version>4.7</jsqlparser.version>
 
         <rocksdb-dependency-scope>compile</rocksdb-dependency-scope>
         <lucene-dependency-scope>compile</lucene-dependency-scope>
@@ -842,6 +843,12 @@
                 <scope>${lucene-dependency-scope}</scope>
             </dependency>
 
+            <dependency>
+                <groupId>com.github.jsqlparser</groupId>
+                <artifactId>jsqlparser</artifactId>
+                <version>${jsqlparser.version}</version>
+            </dependency>
+
             <!-- flink dependencies -->
             <dependency>
                 <groupId>org.apache.flink</groupId>


Test patch:
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java
index a90777d5d..8b67cb568 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java
@@ -33,6 +33,7 @@ import org.apache.amoro.server.utils.IcebergTableUtil;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
+import org.apache.amoro.table.TableProperties;
 import org.apache.amoro.table.TableSnapshot;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.StructLike;
@@ -121,6 +122,83 @@ public class TestOptimizingEvaluator extends MixedTablePlanTestBase {
     assertInput(pendingInput, FileInfo.buildFileInfo(dataFiles));
   }
 
+  @Test
+  public void testFragmentFilesWithPartitionFilterTimeStamp() {
+    getMixedTable()
+        .updateProperties()
+        .set(TableProperties.SELF_OPTIMIZING_FILTER, "op_time >= '2022-01-01T12:00:00'")
+        .commit();
+    testFragmentFilesWithPartitionFilterDo(true);
+
+    getMixedTable()
+        .updateProperties()
+        .set(TableProperties.SELF_OPTIMIZING_FILTER, "op_time > '2022-01-01T12:00:00'")
+        .commit();
+    testFragmentFilesWithPartitionFilterDo(false);
+  }
+
+  @Test
+  public void testFragmentFilesWithPartitionFilterInteger() {
+    getMixedTable()
+        .updateProperties()
+        .set(TableProperties.SELF_OPTIMIZING_FILTER, "id > 0")
+        .commit();
+    testFragmentFilesWithPartitionFilterDo(true);
+
+    getMixedTable()
+        .updateProperties()
+        .set(TableProperties.SELF_OPTIMIZING_FILTER, "id > 8")
+        .commit();
+    testFragmentFilesWithPartitionFilterDo(false);
+  }
+
+  @Test
+  public void testFragmentFilesWithPartitionFilterString() {
+    getMixedTable()
+        .updateProperties()
+        .set(TableProperties.SELF_OPTIMIZING_FILTER, "name > '0'")
+        .commit();
+    testFragmentFilesWithPartitionFilterDo(true);
+
+    getMixedTable()
+        .updateProperties()
+        .set(TableProperties.SELF_OPTIMIZING_FILTER, "name > '8'")
+        .commit();
+    testFragmentFilesWithPartitionFilterDo(false);
+  }
+
+  private void testFragmentFilesWithPartitionFilterDo(boolean isNecessary) {
+    closeFullOptimizingInterval();
+    updateBaseHashBucket(1);
+    List<DataFile> dataFiles = Lists.newArrayList();
+    List<Record> newRecords =
+        OptimizingTestHelpers.generateRecord(tableTestHelper(), 1, 4, "2022-01-01T12:00:00");
+    long transactionId = beginTransaction();
+    dataFiles.addAll(
+        OptimizingTestHelpers.appendBase(
+            getMixedTable(),
+            tableTestHelper().writeBaseStore(getMixedTable(), transactionId, newRecords, false)));
+
+    // add more files
+    newRecords =
+        OptimizingTestHelpers.generateRecord(tableTestHelper(), 5, 8, "2022-01-01T12:00:00");
+    transactionId = beginTransaction();
+    dataFiles.addAll(
+        OptimizingTestHelpers.appendBase(
+            getMixedTable(),
+            tableTestHelper().writeBaseStore(getMixedTable(), transactionId, newRecords, false)));
+
+    AbstractOptimizingEvaluator optimizingEvaluator = buildOptimizingEvaluator();
+    if (isNecessary) {
+      Assert.assertTrue(optimizingEvaluator.isNecessary());
+      AbstractOptimizingEvaluator.PendingInput pendingInput =
+          optimizingEvaluator.getOptimizingPendingInput();
+      assertInput(pendingInput, FileInfo.buildFileInfo(dataFiles));
+    } else {
+      Assert.assertFalse(optimizingEvaluator.isNecessary());
+    }
+  }
+
   protected AbstractOptimizingEvaluator buildOptimizingEvaluator() {
     TableSnapshot snapshot = IcebergTableUtil.getSnapshot(getMixedTable(), tableRuntime);
     return IcebergTableUtil.createOptimizingEvaluator(tableRuntime, getMixedTable(), snapshot, 100);
diff --git a/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestExpressionUtil.java b/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestExpressionUtil.java
new file mode 100644
index 000000000..03707da7b
--- /dev/null
+++ b/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestExpressionUtil.java
@@ -0,0 +1,127 @@
+/*
+ * 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.amoro.utils;
+
+import static org.apache.amoro.utils.ExpressionUtil.convertSqlFilterToIcebergExpression;
+
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoField;
+import java.util.ArrayList;
+import java.util.List;
+
+public class TestExpressionUtil {
+  @Test
+  public void testConvertSqlToIcebergExpression() {
+    List<Types.NestedField> fields = new ArrayList<>();
+    fields.add(Types.NestedField.optional(1, "column_a", Types.IntegerType.get()));
+    assertEqualExpressions(
+        Expressions.isNull("column_a"),
+        convertSqlFilterToIcebergExpression("column_a IS NULL", fields));
+    assertEqualExpressions(
+        Expressions.notNull("column_a"),
+        convertSqlFilterToIcebergExpression("column_a IS NOT NULL", fields));
+
+    assertEqualExpressions(
+        Expressions.alwaysTrue(), convertSqlFilterToIcebergExpression("", fields));
+
+    testConvertSqlToIcebergExpressionByType("1", "'1'", Types.StringType.get());
+    testConvertSqlToIcebergExpressionByType(1, "1", Types.IntegerType.get());
+    testConvertSqlToIcebergExpressionByType(1L, "1", Types.LongType.get());
+    testConvertSqlToIcebergExpressionByType(1.15f, "1.15", Types.FloatType.get());
+    testConvertSqlToIcebergExpressionByType(1.15, "1.15", Types.DoubleType.get());
+    testConvertSqlToIcebergExpressionByType(true, "true", Types.BooleanType.get());
+    testConvertSqlToIcebergExpressionByType(false, "false", Types.BooleanType.get());
+
+    long epochDay = Date.valueOf("2022-01-01").toLocalDate().toEpochDay();
+    testConvertSqlToIcebergExpressionByType(epochDay, "DATE '2022-01-01'", Types.DateType.get());
+    testConvertSqlToIcebergExpressionByType(epochDay, "'2022-01-01'", Types.DateType.get());
+
+    long microOfDay = Time.valueOf("12:12:12").toLocalTime().getLong(ChronoField.MICRO_OF_DAY);
+    testConvertSqlToIcebergExpressionByType(microOfDay, "TIME '12:12:12'", Types.TimeType.get());
+    testConvertSqlToIcebergExpressionByType(microOfDay, "'12:12:12'", Types.TimeType.get());
+
+    long epochMicroSecond =
+        Timestamp.valueOf("2022-01-01 12:12:12")
+                .toLocalDateTime()
+                .toEpochSecond(ZoneOffset.ofHours(0))
+            * 1_000_000L;
+    testConvertSqlToIcebergExpressionByType(
+        epochMicroSecond, "TIMESTAMP '2022-01-01T12:12:12'", Types.TimestampType.withoutZone());
+    testConvertSqlToIcebergExpressionByType(
+        epochMicroSecond, "'2022-01-01T12:12:12'", Types.TimestampType.withoutZone());
+  }
+
+  public <T> void testConvertSqlToIcebergExpressionByType(T exprValue, String sqlValue, Type type) {
+    List<Types.NestedField> fields = new ArrayList<>();
+    fields.add(Types.NestedField.optional(1, "column_a", type));
+    fields.add(Types.NestedField.optional(2, "column_b", type));
+
+    assertEqualExpressions(
+        Expressions.equal("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a = " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.notEqual("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a != " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.greaterThan("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a > " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.greaterThanOrEqual("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a >= " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.lessThan("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a < " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.lessThanOrEqual("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a <= " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.in("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a IN (" + sqlValue + ")", fields));
+    assertEqualExpressions(
+        Expressions.notIn("column_a", exprValue),
+        convertSqlFilterToIcebergExpression("column_a NOT IN (" + sqlValue + ")", fields));
+    assertEqualExpressions(
+        Expressions.not(Expressions.equal("column_a", exprValue)),
+        convertSqlFilterToIcebergExpression("NOT column_a = " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.and(
+            Expressions.equal("column_a", exprValue), Expressions.equal("column_b", exprValue)),
+        convertSqlFilterToIcebergExpression(
+            "column_a = " + sqlValue + " AND column_b = " + sqlValue, fields));
+    assertEqualExpressions(
+        Expressions.or(
+            Expressions.equal("column_a", exprValue), Expressions.equal("column_b", exprValue)),
+        convertSqlFilterToIcebergExpression(
+            "column_a = " + sqlValue + " OR column_b = " + sqlValue, fields));
+  }
+
+  private void assertEqualExpressions(Expression exp1, Expression exp2) {
+    Assert.assertEquals(exp1.toString(), exp2.toString());
+  }
+}