log record batch filter push down

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/record/DefaultLogRecordBatch.java

@@ -33,13 +33,13 @@
 
 import java.nio.ByteBuffer;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 
 import static com.alibaba.fluss.record.LogRecordBatchFormat.BASE_OFFSET_OFFSET;
 import static com.alibaba.fluss.record.LogRecordBatchFormat.LENGTH_OFFSET;
 import static com.alibaba.fluss.record.LogRecordBatchFormat.LOG_OVERHEAD;
 import static com.alibaba.fluss.record.LogRecordBatchFormat.MAGIC_OFFSET;
 import static com.alibaba.fluss.record.LogRecordBatchFormat.NO_LEADER_EPOCH;
-import static com.alibaba.fluss.record.LogRecordBatchFormat.arrowChangeTypeOffset;
 import static com.alibaba.fluss.record.LogRecordBatchFormat.attributeOffset;
 import static com.alibaba.fluss.record.LogRecordBatchFormat.batchSequenceOffset;
 import static com.alibaba.fluss.record.LogRecordBatchFormat.commitTimestampOffset;
@@ -282,7 +282,7 @@ private CloseableIterator<LogRecord> columnRecordIterator(
         if (isAppendOnly) {
             // append only batch, no change type vector,
             // the start of the arrow data is the beginning of the batch records
-            int recordBatchHeaderSize = recordBatchHeaderSize(magic);
+            int recordBatchHeaderSize = getActualRecordBatchHeaderSize(magic);
             int arrowOffset = position + recordBatchHeaderSize;
             int arrowLength = sizeInBytes() - recordBatchHeaderSize;
             ArrowReader reader =
@@ -297,12 +297,14 @@ protected ChangeType getChangeType(int rowId) {
         } else {
             // with change type, decode the change type vector first,
             // the arrow data starts after the change type vector
-            int changeTypeOffset = position + arrowChangeTypeOffset(magic);
+            int changeTypeOffset = position + getActualArrowChangeTypeOffset(magic);
             ChangeTypeVector changeTypeVector =
                     new ChangeTypeVector(segment, changeTypeOffset, getRecordCount());
             int arrowOffset = changeTypeOffset + changeTypeVector.sizeInBytes();
             int arrowLength =
-                    sizeInBytes() - arrowChangeTypeOffset(magic) - changeTypeVector.sizeInBytes();
+                    sizeInBytes()
+                            - getActualArrowChangeTypeOffset(magic)
+                            - changeTypeVector.sizeInBytes();
             ArrowReader reader =
                     ArrowUtils.createArrowReader(
                             segment, arrowOffset, arrowLength, root, allocator, rowType);
@@ -410,4 +412,98 @@ public void remove() {
             throw new UnsupportedOperationException();
         }
     }
+
+    @Override
+    public Optional<LogRecordBatchStatistics> getStatistics(ReadContext context) {
+        if (context == null) {
+            return Optional.empty();
+        }
+
+        byte magic = magic();
+        if (magic < LogRecordBatchFormat.LOG_MAGIC_VALUE_V2) {
+            // Statistics are only available in V2 and later
+            return Optional.empty();
+        }
+
+        // Check if statistics flag is set
+        byte attributes = attributes();
+        if ((attributes & LogRecordBatchFormat.STATISTICS_FLAG_MASK) == 0) {
+            // No statistics available
+            return Optional.empty();
+        }
+
+        try {
+            // Get row type from context
+            RowType rowType = context.getRowType(schemaId());
+            if (rowType == null) {
+                return Optional.empty();
+            }
+
+            // Read statistics length
+            int statisticsLength =
+                    segment.getInt(position + LogRecordBatchFormat.statisticsLengthOffset(magic));
+            if (statisticsLength <= 0) {
+                return Optional.empty();
+            }
+
+            // Read statistics data
+            int statisticsDataOffset = position + LogRecordBatchFormat.statisticsDataOffset(magic);
+            byte[] statisticsData = new byte[statisticsLength];
+            segment.get(statisticsDataOffset, statisticsData, 0, statisticsLength);
+
+            // Deserialize statistics using the row type from context
+            LogRecordBatchStatistics statistics =
+                    LogRecordBatchStatisticsSerializer.deserialize(statisticsData, rowType);
+            return Optional.ofNullable(statistics);
+        } catch (Exception e) {
+            // If reading or deserializing statistics fails, return empty
+            return Optional.empty();
+        }
+    }
+
+    /** Get the actual record batch header size including statistics for V2 format. */
+    private int getActualRecordBatchHeaderSize(byte magic) {
+        if (magic < LogRecordBatchFormat.LOG_MAGIC_VALUE_V2) {
+            return LogRecordBatchFormat.recordBatchHeaderSize(magic);
+        }
+
+        // For V2, we need to add statistics length
+        int statisticsLength = 0;
+        byte attributes = attributes();
+        if ((attributes & LogRecordBatchFormat.STATISTICS_FLAG_MASK) != 0) {
+            try {
+                statisticsLength =
+                        segment.getInt(
+                                position + LogRecordBatchFormat.statisticsLengthOffset(magic));
+            } catch (Exception e) {
+                // If reading statistics length fails, use 0
+                statisticsLength = 0;
+            }
+        }
+
+        return LogRecordBatchFormat.recordBatchHeaderSizeWithStats(magic, statisticsLength);
+    }
+
+    /** Get the actual arrow change type offset including statistics for V2 format. */
+    private int getActualArrowChangeTypeOffset(byte magic) {
+        if (magic < LogRecordBatchFormat.LOG_MAGIC_VALUE_V2) {
+            return LogRecordBatchFormat.arrowChangeTypeOffset(magic);
+        }
+
+        // For V2, we need to add statistics length
+        int statisticsLength = 0;
+        byte attributes = attributes();
+        if ((attributes & LogRecordBatchFormat.STATISTICS_FLAG_MASK) != 0) {
+            try {
+                statisticsLength =
+                        segment.getInt(
+                                position + LogRecordBatchFormat.statisticsLengthOffset(magic));
+            } catch (Exception e) {
+                // If reading statistics length fails, use 0
+                statisticsLength = 0;
+            }
+        }
+
+        return LogRecordBatchFormat.arrowChangeTypeOffsetWithStats(magic, statisticsLength);
+    }
 }

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

@@ -0,0 +1,63 @@
+/*
+ * 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 com.alibaba.fluss.record;
+
+import com.alibaba.fluss.row.InternalArray;
+import com.alibaba.fluss.row.InternalRow;
+
+/** Default implementation of {@link LogRecordBatchStatistics}. */
+public class DefaultLogRecordBatchStatistics implements LogRecordBatchStatistics {
+
+    private final InternalRow minValues;
+    private final InternalRow maxValues;
+    private final InternalArray nullCounts;
+
+    public DefaultLogRecordBatchStatistics(
+            InternalRow minValues, InternalRow maxValues, InternalArray nullCounts) {
+        this.minValues = minValues;
+        this.maxValues = maxValues;
+        this.nullCounts = nullCounts;
+    }
+
+    @Override
+    public InternalRow getMinValues() {
+        return minValues;
+    }
+
+    @Override
+    public InternalRow getMaxValues() {
+        return maxValues;
+    }
+
+    @Override
+    public InternalArray getNullCounts() {
+        return nullCounts;
+    }
+
+    @Override
+    public String toString() {
+        return "DefaultLogRecordBatchStatistics{"
+                + ", minValues="
+                + minValues
+                + ", maxValues="
+                + maxValues
+                + ", nullCounts="
+                + nullCounts
+                + '}';
+    }
+}