filter

Author: platinumhamburg

.gitignore

@@ -42,3 +42,6 @@ website/versioned_docs
 website/versioned_sidebars
 website/versions.json
 website/pnpm-lock.yaml
+
+# jenv
+.java-version

fluss-client/src/main/java/com/alibaba/fluss/client/table/scanner/Scan.java

@@ -21,6 +21,7 @@
 import com.alibaba.fluss.client.table.scanner.batch.BatchScanner;
 import com.alibaba.fluss.client.table.scanner.log.LogScanner;
 import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.predicate.Predicate;
 
 import javax.annotation.Nullable;
 
@@ -65,6 +66,13 @@ public interface Scan {
      */
     LogScanner createLogScanner();
 
+    /**
+     * Creates a {@link LogScanner} to continuously read log data for this scan with filter.
+     *
+     * <p>Note: this API doesn't support pre-configured with {@link #limit(int)}.
+     */
+    LogScanner createLogScanner(Predicate recordBatchFilter);
+
     /**
      * Creates a {@link BatchScanner} to read current data in the given table bucket for this scan.
      *

fluss-client/src/main/java/com/alibaba/fluss/client/table/scanner/TableScan.java

@@ -29,6 +29,7 @@
 import com.alibaba.fluss.exception.FlussRuntimeException;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TableInfo;
+import com.alibaba.fluss.predicate.Predicate;
 import com.alibaba.fluss.types.RowType;
 
 import javax.annotation.Nullable;
@@ -43,6 +44,7 @@ public class TableScan implements Scan {
 
     /** The projected fields to do projection. No projection if is null. */
     @Nullable private final int[] projectedColumns;
+
     /** The limited row number to read. No limit if is null. */
     @Nullable private final Integer limit;
 
@@ -98,7 +100,24 @@ public LogScanner createLogScanner() {
                 conn.getMetadataUpdater(),
                 conn.getClientMetricGroup(),
                 conn.getOrCreateRemoteFileDownloader(),
-                projectedColumns);
+                projectedColumns,
+                null);
+    }
+
+    @Override
+    public LogScanner createLogScanner(Predicate recordBatchFilter) {
+        if (limit != null) {
+            throw new UnsupportedOperationException("LogScanner doesn't support limit pushdown.");
+        }
+        return new LogScannerImpl(
+                conn.getConfiguration(),
+                tableInfo,
+                conn.getRpcClient(),
+                conn.getMetadataUpdater(),
+                conn.getClientMetricGroup(),
+                conn.getOrCreateRemoteFileDownloader(),
+                projectedColumns,
+                recordBatchFilter);
     }
 
     @Override

fluss-client/src/main/java/com/alibaba/fluss/client/table/scanner/log/LogFetcher.java

@@ -35,6 +35,7 @@
 import com.alibaba.fluss.metadata.TableInfo;
 import com.alibaba.fluss.metadata.TablePartition;
 import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.predicate.Predicate;
 import com.alibaba.fluss.record.LogRecordReadContext;
 import com.alibaba.fluss.record.LogRecords;
 import com.alibaba.fluss.record.MemoryLogRecords;
@@ -51,6 +52,7 @@
 import com.alibaba.fluss.rpc.messages.PbFetchLogRespForBucket;
 import com.alibaba.fluss.rpc.messages.PbFetchLogRespForTable;
 import com.alibaba.fluss.rpc.protocol.Errors;
+import com.alibaba.fluss.rpc.util.PredicateMessageUtils;
 import com.alibaba.fluss.utils.IOUtils;
 import com.alibaba.fluss.utils.Projection;
 
@@ -91,6 +93,7 @@ public class LogFetcher implements Closeable {
     //  bytes from remote file.
     private final LogRecordReadContext remoteReadContext;
     @Nullable private final Projection projection;
+    @Nullable private final Predicate recordBatchFilter;
     private final RpcClient rpcClient;
     private final int maxFetchBytes;
     private final int maxBucketFetchBytes;
@@ -114,6 +117,7 @@ public class LogFetcher implements Closeable {
     public LogFetcher(
             TableInfo tableInfo,
             @Nullable Projection projection,
+            @Nullable Predicate recordBatchFilter,
             RpcClient rpcClient,
             LogScannerStatus logScannerStatus,
             Configuration conf,
@@ -126,6 +130,7 @@ public LogFetcher(
         this.remoteReadContext =
                 LogRecordReadContext.createReadContext(tableInfo, true, projection);
         this.projection = projection;
+        this.recordBatchFilter = recordBatchFilter;
         this.rpcClient = rpcClient;
         this.logScannerStatus = logScannerStatus;
         this.maxFetchBytes =
@@ -461,6 +466,10 @@ private Map<Integer, FetchLogRequest> prepareFetchLogRequests() {
                         } else {
                             reqForTable.setProjectionPushdownEnabled(false);
                         }
+                        if (null != recordBatchFilter) {
+                            reqForTable.setRecordBatchFilter(
+                                    PredicateMessageUtils.toPbPredicate(recordBatchFilter));
+                        }
                         reqForTable.addAllBucketsReqs(reqForBuckets);
                         fetchLogRequest.addAllTablesReqs(Collections.singletonList(reqForTable));
                         fetchLogRequests.put(leaderId, fetchLogRequest);

fluss-client/src/main/java/com/alibaba/fluss/client/table/scanner/log/LogScannerImpl.java

@@ -27,6 +27,7 @@
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TableInfo;
 import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.predicate.Predicate;
 import com.alibaba.fluss.rpc.RpcClient;
 import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
 import com.alibaba.fluss.types.RowType;
@@ -84,7 +85,8 @@ public LogScannerImpl(
             MetadataUpdater metadataUpdater,
             ClientMetricGroup clientMetricGroup,
             RemoteFileDownloader remoteFileDownloader,
-            @Nullable int[] projectedFields) {
+            @Nullable int[] projectedFields,
+            @Nullable Predicate recordBatchFilter) {
         this.tablePath = tableInfo.getTablePath();
         this.tableId = tableInfo.getTableId();
         this.isPartitionedTable = tableInfo.isPartitioned();
@@ -98,6 +100,7 @@ public LogScannerImpl(
                 new LogFetcher(
                         tableInfo,
                         projection,
+                        recordBatchFilter,
                         rpcClient,
                         logScannerStatus,
                         conf,

fluss-client/src/test/java/com/alibaba/fluss/client/table/scanner/log/LogFetcherTest.java

@@ -81,6 +81,7 @@ protected void setup() throws Exception {
                 new LogFetcher(
                         DATA1_TABLE_INFO,
                         null,
+                        null,
                         rpcClient,
                         logScannerStatus,
                         clientConf,
@@ -158,6 +159,7 @@ void testFetchWhenDestinationIsNullInMetadata() throws Exception {
                 new LogFetcher(
                         DATA1_TABLE_INFO,
                         null,
+                        null,
                         rpcClient,
                         logScannerStatus,
                         clientConf,

fluss-common/src/main/java/com/alibaba/fluss/predicate/And.java

@@ -17,6 +17,7 @@
 
 package com.alibaba.fluss.predicate;
 
+import com.alibaba.fluss.row.InternalArray;
 import com.alibaba.fluss.row.InternalRow;
 
 import java.util.ArrayList;
@@ -46,6 +47,21 @@ public boolean test(InternalRow row, List<Predicate> children) {
         return true;
     }
 
+    @Override
+    public boolean test(
+            long rowCount,
+            InternalRow minValues,
+            InternalRow maxValues,
+            InternalArray nullCounts,
+            List<Predicate> children) {
+        for (Predicate child : children) {
+            if (!child.test(rowCount, minValues, maxValues, nullCounts)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     @Override
     public Optional<Predicate> negate(List<Predicate> children) {
         List<Predicate> negatedChildren = new ArrayList<>();

fluss-common/src/main/java/com/alibaba/fluss/predicate/CompoundPredicate.java

@@ -17,6 +17,7 @@
 
 package com.alibaba.fluss.predicate;
 
+import com.alibaba.fluss.row.InternalArray;
 import com.alibaba.fluss.row.InternalRow;
 
 import java.io.Serializable;
@@ -55,6 +56,12 @@ public boolean test(InternalRow row) {
         return function.test(row, children);
     }
 
+    @Override
+    public boolean test(
+            long rowCount, InternalRow minValues, InternalRow maxValues, InternalArray nullCounts) {
+        return function.test(rowCount, minValues, maxValues, nullCounts, children);
+    }
+
     @Override
     public Optional<Predicate> negate() {
         return function.negate(children);
@@ -89,6 +96,13 @@ public abstract static class Function implements Serializable {
 
         public abstract boolean test(InternalRow row, List<Predicate> children);
 
+        public abstract boolean test(
+                long rowCount,
+                InternalRow minValues,
+                InternalRow maxValues,
+                InternalArray nullCounts,
+                List<Predicate> children);
+
         public abstract Optional<Predicate> negate(List<Predicate> children);
 
         public abstract <T> T visit(FunctionVisitor<T> visitor, List<T> children);

fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafPredicate.java

@@ -17,6 +17,7 @@
 
 package com.alibaba.fluss.predicate;
 
+import com.alibaba.fluss.row.InternalArray;
 import com.alibaba.fluss.row.InternalRow;
 import com.alibaba.fluss.types.DataType;
 import com.alibaba.fluss.types.DecimalType;
@@ -89,6 +90,23 @@ public boolean test(InternalRow row) {
         return function.test(type, get(row, fieldIndex, type), literals);
     }
 
+    @Override
+    public boolean test(
+            long rowCount, InternalRow minValues, InternalRow maxValues, InternalArray nullCounts) {
+        Object min = get(minValues, fieldIndex, type);
+        Object max = get(maxValues, fieldIndex, type);
+        Long nullCount = nullCounts.isNullAt(fieldIndex) ? null : nullCounts.getLong(fieldIndex);
+        if (nullCount == null || rowCount != nullCount) {
+            // not all null
+            // min or max is null
+            // unknown stats
+            if (min == null || max == null) {
+                return true;
+            }
+        }
+        return function.test(type, rowCount, min, max, nullCount, literals);
+    }
+
     @Override
     public Optional<Predicate> negate() {
         return function.negate()

fluss-common/src/main/java/com/alibaba/fluss/predicate/Or.java

@@ -17,6 +17,7 @@
 
 package com.alibaba.fluss.predicate;
 
+import com.alibaba.fluss.row.InternalArray;
 import com.alibaba.fluss.row.InternalRow;
 
 import java.util.ArrayList;
@@ -46,6 +47,21 @@ public boolean test(InternalRow row, List<Predicate> children) {
         return false;
     }
 
+    @Override
+    public boolean test(
+            long rowCount,
+            InternalRow minValues,
+            InternalRow maxValues,
+            InternalArray nullCounts,
+            List<Predicate> children) {
+        for (Predicate child : children) {
+            if (child.test(rowCount, minValues, maxValues, nullCounts)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     public Optional<Predicate> negate(List<Predicate> children) {
         List<Predicate> negatedChildren = new ArrayList<>();