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;
 
@@ -51,6 +52,8 @@ public interface Scan {
      */
     Scan project(List<String> projectedColumnNames);
 
+    Scan lossyFilter(Predicate loosyFilter);
+
     /**
      * Returns a new scan from this that will read the given limited row number.
      *

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,27 +44,32 @@ public class TableScan implements Scan {
 
     /** The projected fields to do projection. No projection if is null. */
     @Nullable private final int[] projectedColumns;
+
+    @Nullable private final Predicate loosyFilter;
+
     /** The limited row number to read. No limit if is null. */
     @Nullable private final Integer limit;
 
     public TableScan(FlussConnection conn, TableInfo tableInfo) {
-        this(conn, tableInfo, null, null);
+        this(conn, tableInfo, null, null, null);
     }
 
     private TableScan(
             FlussConnection conn,
             TableInfo tableInfo,
             @Nullable int[] projectedColumns,
+            @Nullable Predicate loosyFilter,
             @Nullable Integer limit) {
         this.conn = conn;
         this.tableInfo = tableInfo;
         this.projectedColumns = projectedColumns;
+        this.loosyFilter = loosyFilter;
         this.limit = limit;
     }
 
     @Override
     public Scan project(@Nullable int[] projectedColumns) {
-        return new TableScan(conn, tableInfo, projectedColumns, limit);
+        return new TableScan(conn, tableInfo, projectedColumns, loosyFilter, limit);
     }
 
     @Override
@@ -78,12 +84,17 @@ public Scan project(List<String> projectedColumnNames) {
             }
             columnIndexes[i] = index;
         }
-        return new TableScan(conn, tableInfo, columnIndexes, limit);
+        return new TableScan(conn, tableInfo, columnIndexes, loosyFilter, limit);
+    }
+
+    @Override
+    public Scan lossyFilter(Predicate loosyFilter) {
+        return new TableScan(conn, tableInfo, projectedColumns, loosyFilter, limit);
     }
 
     @Override
     public Scan limit(int rowNumber) {
-        return new TableScan(conn, tableInfo, projectedColumns, rowNumber);
+        return new TableScan(conn, tableInfo, projectedColumns, loosyFilter, rowNumber);
     }
 
     @Override
@@ -98,7 +109,8 @@ public LogScanner createLogScanner() {
                 conn.getMetadataUpdater(),
                 conn.getClientMetricGroup(),
                 conn.getOrCreateRemoteFileDownloader(),
-                projectedColumns);
+                projectedColumns,
+                loosyFilter);
     }
 
     @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;
@@ -91,6 +92,7 @@ public class LogFetcher implements Closeable {
     //  bytes from remote file.
     private final LogRecordReadContext remoteReadContext;
     @Nullable private final Projection projection;
+    @Nullable private final Predicate loosyFilter;
     private final RpcClient rpcClient;
     private final int maxFetchBytes;
     private final int maxBucketFetchBytes;
@@ -114,6 +116,7 @@ public class LogFetcher implements Closeable {
     public LogFetcher(
             TableInfo tableInfo,
             @Nullable Projection projection,
+            @Nullable Predicate loosyFilter,
             RpcClient rpcClient,
             LogScannerStatus logScannerStatus,
             Configuration conf,
@@ -126,6 +129,7 @@ public LogFetcher(
         this.remoteReadContext =
                 LogRecordReadContext.createReadContext(tableInfo, true, projection);
         this.projection = projection;
+        this.loosyFilter = loosyFilter;
         this.rpcClient = rpcClient;
         this.logScannerStatus = logScannerStatus;
         this.maxFetchBytes =

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 loosyFilter) {
         this.tablePath = tableInfo.getTablePath();
         this.tableId = tableInfo.getTableId();
         this.isPartitionedTable = tableInfo.isPartitioned();
@@ -98,6 +100,7 @@ public LogScannerImpl(
                 new LogFetcher(
                         tableInfo,
                         projection,
+                        loosyFilter,
                         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/record/DefaultLogRecordBatch.java

@@ -34,6 +34,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 
 /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
@@ -434,4 +435,9 @@ public void remove() {
             throw new UnsupportedOperationException();
         }
     }
+
+    @Override
+    public Optional<LogRecordBatchStatistics> getStatistics() {
+        return Optional.empty();
+    }
 }

fluss-common/src/main/java/com/alibaba/fluss/record/FileLogInputStream.java

@@ -27,6 +27,7 @@
 import java.nio.ByteOrder;
 import java.nio.channels.FileChannel;
 import java.util.Objects;
+import java.util.Optional;
 
 import static com.alibaba.fluss.record.DefaultLogRecordBatch.BASE_OFFSET_OFFSET;
 import static com.alibaba.fluss.record.DefaultLogRecordBatch.LENGTH_OFFSET;
@@ -271,5 +272,10 @@ public String toString() {
                     + batchSize
                     + ")";
         }
+
+        @Override
+        public Optional<LogRecordBatchStatistics> getStatistics() {
+            return Optional.empty();
+        }
     }
 }

fluss-common/src/main/java/com/alibaba/fluss/record/LogRecordBatch.java

@@ -25,6 +25,7 @@
 import com.alibaba.fluss.utils.CloseableIterator;
 
 import java.util.Iterator;
+import java.util.Optional;
 
 /**
  * A record batch is a container for {@link LogRecord LogRecords}.
@@ -52,6 +53,13 @@ public interface LogRecordBatch {
      */
     boolean isValid();
 
+    /**
+     * Get the statistics of this record batch.
+     *
+     * @return
+     */
+    Optional<LogRecordBatchStatistics> getStatistics();
+
     /** Raise an exception if the checksum is not valid. */
     void ensureValid();
 

fluss-common/src/main/java/com/alibaba/fluss/record/LogRecordBatchStatistics.java

@@ -0,0 +1,4 @@
+package com.alibaba.fluss.record;
+
+/** Statistics infomation of {@link LogRecordBatch LogRecordBatch}. */
+public class LogRecordBatchStatistics {}