server-side implementation

Author: polyzos

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1120,6 +1120,13 @@ public class ConfigOptions {
                                     + CLIENT_SCANNER_LOG_FETCH_WAIT_MAX_TIME.key()
                                     + " time to return.");
 
+    public static final ConfigOption<Integer> CLIENT_SCANNER_KV_FETCH_MAX_BYTES =
+            key("client.scanner.kv.fetch.max-bytes")
+                    .intType()
+                    .defaultValue(4 * 1024 * 1024)
+                    .withDescription(
+                            "Max bytes per streaming KV scan batch (ScanKvRequest.batch_size_bytes).");
+
     public static final ConfigOption<Integer> CLIENT_LOOKUP_QUEUE_SIZE =
             key("client.lookup.queue-size")
                     .intType()

fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java

@@ -272,6 +272,10 @@ public void append(byte[] valueBytes) throws IOException {
             currentRecordNumber++;
         }
 
+        public int sizeInBytes() {
+            return sizeInBytes;
+        }
+
         public DefaultValueRecordBatch build() throws IOException {
             writeBatchHeader();
             MemorySegment segment = outputView.getMemorySegment();

fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java

@@ -0,0 +1,112 @@
+/*
+ * 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.fluss.server.kv.scan;
+
+import org.rocksdb.ReadOptions;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksIterator;
+import org.rocksdb.Snapshot;
+
+/**
+ * Holds server-side state for a single streaming KV scan.
+ *
+ * <p>A scanner is bound to exactly one bucket (table_id, partition_id, bucket_id) and owns:
+ *
+ * <ul>
+ *   <li>a RocksDB {@link Snapshot} providing a consistent point-in-time view
+ *   <li>a {@link RocksIterator} iterating over that snapshot
+ *   <li>sequencing state (call_seq_id) to detect missed/out-of-order batches
+ *   <li>a scan boundary {@code logOffset} captured at scanner creation
+ * </ul>
+ *
+ * <p>Lifecycle: created on first ScanKvRequest; closed when exhausted, explicitly closed, or
+ * expired by {@link ScannerManager}.
+ */
+public class ScannerContext implements AutoCloseable {
+    private String scannerId;
+    private RocksDB db;
+    private Snapshot snapshot;
+    private ReadOptions readOptions;
+    private RocksIterator iterator;
+
+    private long logOffset;
+    private long rowLimit; // -1 means unlimited
+
+    private long emittedRows;
+    private int expectedCallSeqId;
+    private volatile long lastAccessNanos;
+
+    ScannerContext(String scannerId, RocksDB db, long logOffset, long rowLimit, long nowNanos) {
+        this.scannerId = scannerId;
+        this.db = db;
+        this.logOffset = logOffset;
+        this.rowLimit = rowLimit;
+
+        this.snapshot = db.getSnapshot();
+        this.readOptions = new ReadOptions().setSnapshot(snapshot);
+        this.iterator = db.newIterator(readOptions);
+        this.iterator.seekToFirst();
+
+        this.lastAccessNanos = nowNanos;
+        this.emittedRows = 0L;
+        this.expectedCallSeqId = 0;
+    }
+
+    public String getScannerId() {
+        return scannerId;
+    }
+
+    public int getExpectedCallSeqId() {
+        return expectedCallSeqId;
+    }
+
+    public RocksIterator getIterator() {
+        return iterator;
+    }
+
+    public void advanceSeq() {
+        expectedCallSeqId++;
+    }
+
+    public void touch(long nowNanos) {
+        lastAccessNanos = nowNanos;
+    }
+
+    public boolean limitReached() {
+        return rowLimit >= 0 && emittedRows >= rowLimit;
+    }
+
+    public void incrementEmitted() {
+        emittedRows++;
+    }
+
+    public long getLogOffset() {
+        return logOffset;
+    }
+
+    public long getLastAccessNanos() {
+        return lastAccessNanos;
+    }
+
+    @Override
+    public void close() {
+        iterator.close();
+        readOptions.close();
+        db.releaseSnapshot(snapshot);
+    }
+}

fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerManager.java

@@ -0,0 +1,103 @@
+/*
+ * 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.fluss.server.kv.scan;
+
+import org.apache.fluss.utils.MapUtils;
+
+import org.rocksdb.RocksDB;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Manages server-side KV scan scanners.
+ *
+ * <p>Responsibilities:
+ *
+ * <ul>
+ *   <li>create & lookup scanners by scannerId
+ *   <li>expire idle scanners based on {@code server.scanner.ttl}
+ *   <li>release RocksDB snapshot resources on removal
+ * </ul>
+ *
+ * <p>Thread-safety: all public methods are thread-safe. Each scanner itself is single-threaded by
+ * protocol (call_seq_id ordering), enforced at service layer.
+ */
+public class ScannerManager implements AutoCloseable {
+    private final Map<String, ScannerContext> scanners = MapUtils.newConcurrentHashMap();
+    private final long ttlNanos;
+    private final ScheduledExecutorService cleanerThread;
+
+    public ScannerManager() {
+        this.ttlNanos = Duration.ofMinutes(5).toNanos();
+        this.cleanerThread =
+                Executors.newSingleThreadScheduledExecutor(
+                        r -> {
+                            Thread t = new Thread(r, "scanner-cleaner");
+                            t.setDaemon(true);
+                            return t;
+                        });
+
+        // expires scanners periodically
+        cleanerThread.scheduleAtFixedRate(this::cleanerExpired, 30, 30, TimeUnit.SECONDS);
+    }
+
+    public ScannerContext create(String id, RocksDB db, long logOffset, long rowLimit) {
+        long now = System.nanoTime();
+        ScannerContext ctx = new ScannerContext(id, db, logOffset, rowLimit, now);
+        scanners.put(id, ctx);
+        return ctx;
+    }
+
+    public ScannerContext get(String id) {
+        return scanners.get(id);
+    }
+
+    public void remove(String id) {
+        ScannerContext ctx = scanners.remove(id);
+        if (ctx != null) {
+            ctx.close();
+        }
+    }
+
+    void cleanerExpired() {
+        long now = System.nanoTime();
+        for (var entry : scanners.entrySet()) {
+            if (now - entry.getValue().getLastAccessNanos() > ttlNanos) {
+                remove(entry.getKey());
+            }
+        }
+    }
+
+    @Override
+    public void close() {
+        cleanerThread.shutdownNow();
+        scanners.forEach(
+                (id, ctx) -> {
+                    try {
+                        ctx.close();
+                    } catch (Throwable ignored) {
+                        // best-effort shutdown
+                    }
+                });
+        scanners.clear();
+    }
+}