Draft version of Row filter purge minion task

pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java

@@ -312,4 +312,31 @@ public static class UpsertCompactMergeTask {
 
     public static final String MERGED_SEGMENTS_ZK_SUFFIX = ".mergedSegments";
   }
+
+  // Purges rows inside segment that match filter function.
+  public static class RowFilterPurgeTask {
+    public static final String TASK_TYPE = "RowFilterPurgeTask";
+
+    /**
+     * Task-config key: Groovy/expression filter function applied row-by-row.
+     * Rows for which the function evaluates to {@code true} are removed from the segment.
+     * This is intentionally separate from {@code ingestionConfig.filterConfig.filterFunction} so
+     * that the filter only affects the purge operation and does not alter regular ingestion behaviour
+     * of the table.
+     * <p>Example: {@code "Groovy({status == 'DELETED'}, status)"}
+     */
+    public static final String FILTER_FUNCTION_KEY = "filterFunction";
+
+    /**
+     * Task-config key (boolean, default {@code false}): when {@code true} a row that throws during
+     * filter evaluation is silently marked incomplete instead of failing the task.
+     */
+    public static final String FILTER_CONTINUE_ON_ERROR_KEY = "filterContinueOnError";
+
+    /** Custom-property key reported in SegmentConversionResult: number of rows retained in the segment. */
+    public static final String NUM_ROWS_RETAINED_KEY = "numRowsRetained";
+
+    /** Custom-property key reported in SegmentConversionResult: number of rows dropped by the filter. */
+    public static final String NUM_ROWS_FILTERED_KEY = "numRowsFiltered";
+  }
 }

pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/rowfilterpurge/RowFilterPurgeTaskExecutor.java

@@ -0,0 +1,118 @@
+/**
+ * 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.pinot.plugin.minion.tasks.rowfilterpurge;
+
+import com.google.common.base.Preconditions;
+import java.io.File;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadataCustomMapModifier;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.minion.SegmentPurger;
+import org.apache.pinot.plugin.minion.tasks.BaseSingleSegmentConversionExecutor;
+import org.apache.pinot.plugin.minion.tasks.SegmentConversionResult;
+import org.apache.pinot.plugin.minion.tasks.purge.PurgeTaskExecutor;
+import org.apache.pinot.segment.local.function.FunctionEvaluator;
+import org.apache.pinot.segment.local.function.FunctionEvaluatorFactory;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class RowFilterPurgeTaskExecutor extends BaseSingleSegmentConversionExecutor {
+  private static final Logger LOGGER = LoggerFactory.getLogger(RowFilterPurgeTaskExecutor.class);
+
+  @Override
+  protected SegmentConversionResult convert(PinotTaskConfig pinotTaskConfig, File indexDir, File workingDir)
+      throws Exception {
+    Map<String, String> configs = pinotTaskConfig.getConfigs();
+    String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY);
+    String segmentName = configs.get(MinionConstants.SEGMENT_NAME_KEY);
+
+    TableConfig tableConfig = getTableConfig(tableNameWithType);
+    Schema schema = getSchema(tableNameWithType);
+
+    String filterFunction = configs.get(MinionConstants.RowFilterPurgeTask.FILTER_FUNCTION_KEY);
+    Preconditions.checkArgument(filterFunction != null, "'%s' is required",
+        MinionConstants.RowFilterPurgeTask.FILTER_FUNCTION_KEY);
+    boolean continueOnError = Boolean.parseBoolean(
+        configs.getOrDefault(MinionConstants.RowFilterPurgeTask.FILTER_CONTINUE_ON_ERROR_KEY, "false"));
+
+    FunctionEvaluator filterEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction);
+
+    LOGGER.info("RowFilterPurgeTask: purging segment {} of table {}", segmentName, tableNameWithType);
+    _eventObserver.notifyProgress(pinotTaskConfig,
+        "Purging segment: " + segmentName + " of table: " + tableNameWithType);
+
+    SegmentPurger.RecordPurger recordPurger = row -> {
+      try {
+        return Boolean.TRUE.equals(filterEvaluator.evaluate(row));
+      } catch (Exception e) {
+        if (continueOnError) {
+          LOGGER.debug("Ignoring filter evaluation error for row, treating as retained. Error: {}", e.getMessage());
+          return false;
+        }
+        throw new RuntimeException(
+            "Filter evaluation failed. Set filterContinueOnError=true to skip bad rows. Error: " + e.getMessage(), e);
+      }
+    };
+
+    SegmentPurger segmentPurger =
+        new SegmentPurger(indexDir, workingDir, tableConfig, schema, recordPurger, null, null);
+    File purgedSegmentFile = segmentPurger.purgeSegment();
+
+    int numFiltered = segmentPurger.getNumRecordsPurged();
+    int numRetained = new SegmentMetadataImpl(indexDir).getTotalDocs() - numFiltered;
+
+    LOGGER.info("RowFilterPurgeTask: finished segment {}, rows retained: {}, rows filtered: {}",
+        segmentName, numRetained, numFiltered);
+
+    if (purgedSegmentFile == null) {
+      // Nothing was purged; return original so the executor still stamps ZK.
+      return new SegmentConversionResult.Builder()
+          .setFile(indexDir)
+          .setTableNameWithType(tableNameWithType)
+          .setSegmentName(segmentName)
+          .setCustomProperty(MinionConstants.RowFilterPurgeTask.NUM_ROWS_RETAINED_KEY, numRetained)
+          .setCustomProperty(PurgeTaskExecutor.NUM_RECORDS_PURGED_KEY, numFiltered)
+          .build();
+    }
+
+    return new SegmentConversionResult.Builder()
+        .setFile(purgedSegmentFile)
+        .setTableNameWithType(tableNameWithType)
+        .setSegmentName(segmentName)
+        .setCustomProperty(MinionConstants.RowFilterPurgeTask.NUM_ROWS_RETAINED_KEY, numRetained)
+        .setCustomProperty(MinionConstants.RowFilterPurgeTask.NUM_ROWS_FILTERED_KEY, numFiltered)
+        .build();
+  }
+
+  @Override
+  protected SegmentZKMetadataCustomMapModifier getSegmentZKMetadataCustomMapModifier(
+      PinotTaskConfig pinotTaskConfig, SegmentConversionResult segmentConversionResult) {
+    return new SegmentZKMetadataCustomMapModifier(
+        SegmentZKMetadataCustomMapModifier.ModifyMode.UPDATE,
+        Collections.singletonMap(
+            MinionConstants.RowFilterPurgeTask.TASK_TYPE + MinionConstants.TASK_TIME_SUFFIX,
+            String.valueOf(System.currentTimeMillis())));
+  }
+}

pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/rowfilterpurge/RowFilterPurgeTaskExecutorFactory.java

@@ -0,0 +1,49 @@
+/**
+ * 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.pinot.plugin.minion.tasks.rowfilterpurge;
+
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.minion.MinionConf;
+import org.apache.pinot.minion.executor.MinionTaskZkMetadataManager;
+import org.apache.pinot.minion.executor.PinotTaskExecutor;
+import org.apache.pinot.minion.executor.PinotTaskExecutorFactory;
+import org.apache.pinot.spi.annotations.minion.TaskExecutorFactory;
+
+
+@TaskExecutorFactory
+public class RowFilterPurgeTaskExecutorFactory implements PinotTaskExecutorFactory {
+
+  @Override
+  public void init(MinionTaskZkMetadataManager zkMetadataManager) {
+  }
+
+  @Override
+  public void init(MinionTaskZkMetadataManager zkMetadataManager, MinionConf minionConf) {
+  }
+
+  @Override
+  public String getTaskType() {
+    return MinionConstants.RowFilterPurgeTask.TASK_TYPE;
+  }
+
+  @Override
+  public PinotTaskExecutor create() {
+    return new RowFilterPurgeTaskExecutor();
+  }
+}