You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/07/23 09:26:27 UTC
(fluss) branch main updated: [client] Add log read preference for remote-first fetch (#3606)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 5cacbb02a [client] Add log read preference for remote-first fetch (#3606)
5cacbb02a is described below
commit 5cacbb02a51c189a7e4514566560ced2c031f656
Author: Junfan Zhang <zu...@apache.org>
AuthorDate: Thu Jul 23 17:26:21 2026 +0800
[client] Add log read preference for remote-first fetch (#3606)
* [client] Add log read preference for remote-first fetch
* rust fix
* add e2e
* Potential fix for pull request finding
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
* fallback to local
* fix
* not leader gate
* wait util all tiered
---------
Co-authored-by: zhangjunfan <zh...@qiyi.com>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../fluss/client/table/scanner/log/LogFetcher.java | 6 +-
.../client/table/scanner/log/LogFetcherTest.java | 64 +++++++++---
.../org/apache/fluss/config/ConfigOptions.java | 8 ++
.../fluss/rpc/protocol/FetchLogReadPreference.java | 52 +++++++++
.../fluss/flink/tiering/source/TieringSource.java | 11 +-
.../fluss/flink/tiering/FlinkTieringTestBase.java | 39 +++++++
.../apache/fluss/flink/tiering/TieringITCase.java | 116 +++++++++++++++++----
.../fluss/flink/tiering/TieringFailoverITCase.java | 8 +-
fluss-rpc/src/main/proto/FlussApi.proto | 3 +-
.../crates/fluss/src/client/table/scanner.rs | 1 +
.../org/apache/fluss/server/log/FetchParams.java | 61 ++++++++++-
.../fluss/server/log/FetchParamsBuilder.java | 10 +-
.../fluss/server/replica/ReplicaManager.java | 56 ++++++++++
.../apache/fluss/server/tablet/TabletService.java | 7 ++
.../server/log/remote/RemoteLogManagerTest.java | 99 ++++++++++++++++++
15 files changed, 489 insertions(+), 52 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
index 2a1fc4361..a5b8ba483 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
@@ -52,6 +52,7 @@ import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket;
import org.apache.fluss.rpc.messages.PbFetchLogRespForTable;
import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.rpc.protocol.Errors;
+import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
import org.apache.fluss.rpc.util.PredicateMessageUtils;
import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.ChunkedAllocationManager;
import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf;
@@ -109,6 +110,7 @@ public class LogFetcher implements Closeable {
private final LogFetchCollector logFetchCollector;
private final ArrowLogFetchCollector arrowLogFetchCollector;
private final RemoteLogDownloader remoteLogDownloader;
+ private final FetchLogReadPreference readPreference;
@GuardedBy("this")
private final Set<Integer> nodesWithPendingFetchRequests;
@@ -166,6 +168,7 @@ public class LogFetcher implements Closeable {
this.arrowLogFetchCollector =
new ArrowLogFetchCollector(tablePath, logScannerStatus, conf, metadataUpdater);
this.scannerMetricGroup = scannerMetricGroup;
+ this.readPreference = conf.get(ConfigOptions.CLIENT_SCANNER_LOG_READ_PREFERENCE);
this.remoteLogDownloader =
new RemoteLogDownloader(tablePath, conf, remoteFileDownloader, scannerMetricGroup);
remoteLogDownloader.start();
@@ -562,7 +565,8 @@ public class LogFetcher implements Closeable {
.setFollowerServerId(-1)
.setMaxBytes(maxFetchBytes)
.setMinBytes(minFetchBytes)
- .setMaxWaitMs(maxFetchWaitMs);
+ .setMaxWaitMs(maxFetchWaitMs)
+ .setReadPreference(readPreference.value());
PbFetchLogReqForTable reqForTable =
new PbFetchLogReqForTable().setTableId(finalTableId);
if (readContext.isProjectionPushDowned()) {
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
index 576eab5da..0a96b840d 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
@@ -23,6 +23,7 @@ import org.apache.fluss.client.metadata.TestingMetadataUpdater;
import org.apache.fluss.client.metrics.TestingScannerMetricGroup;
import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
import org.apache.fluss.cluster.BucketLocation;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.NotLeaderOrFollowerException;
import org.apache.fluss.metadata.PhysicalTablePath;
@@ -32,9 +33,12 @@ import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.messages.FetchLogRequest;
import org.apache.fluss.rpc.messages.FetchLogResponse;
import org.apache.fluss.rpc.protocol.ApiError;
+import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
import org.apache.fluss.server.entity.FetchReqInfo;
import org.apache.fluss.server.tablet.TestTabletServerGateway;
+import org.apache.fluss.utils.IOUtils;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -68,24 +72,29 @@ public class LogFetcherTest {
@BeforeEach
public void setup() {
metadataUpdater = initializeMetadataUpdater();
+ logFetcher = createLogFetcher(new Configuration());
+ }
+
+ @AfterEach
+ public void tearDown() {
+ IOUtils.closeQuietly(logFetcher);
+ }
+
+ private LogFetcher createLogFetcher(Configuration conf) {
ClientSchemaGetter clientSchemaGetter =
new TestingClientSchemaGetter(
- DATA1_TABLE_PATH,
- new SchemaInfo(DATA1_SCHEMA, 0),
- metadataUpdater,
- new Configuration());
+ DATA1_TABLE_PATH, new SchemaInfo(DATA1_SCHEMA, 0), metadataUpdater, conf);
LogScannerStatus logScannerStatus = initializeLogScannerStatus();
- logFetcher =
- new LogFetcher(
- DATA1_TABLE_INFO,
- null,
- null,
- logScannerStatus,
- new Configuration(),
- metadataUpdater,
- TestingScannerMetricGroup.newInstance(),
- new RemoteFileDownloader(1),
- clientSchemaGetter);
+ return new LogFetcher(
+ DATA1_TABLE_INFO,
+ null,
+ null,
+ logScannerStatus,
+ conf,
+ metadataUpdater,
+ TestingScannerMetricGroup.newInstance(),
+ new RemoteFileDownloader(1),
+ clientSchemaGetter);
}
@Test
@@ -111,6 +120,31 @@ public class LogFetcherTest {
assertThat(metadataUpdater.getBucketLocation(tb1)).isNotPresent();
}
+ @Test
+ void testPrepareFetchLogRequestWithReadPreference() throws Exception {
+ Map<Integer, FetchLogRequest> defaultRequestMap = logFetcher.prepareFetchLogRequests();
+ FetchLogRequest defaultRequest = defaultRequestMap.get(1);
+ assertThat(defaultRequest.hasReadPreference()).isTrue();
+ assertThat(defaultRequest.getReadPreference())
+ .isEqualTo(FetchLogReadPreference.LOCAL_FIRST.value());
+
+ Configuration remoteFirstConf = new Configuration();
+ remoteFirstConf.setString(
+ ConfigOptions.CLIENT_SCANNER_LOG_READ_PREFERENCE.key(),
+ FetchLogReadPreference.REMOTE_FIRST.toString());
+ LogFetcher remoteFirstFetcher = createLogFetcher(remoteFirstConf);
+ try {
+ Map<Integer, FetchLogRequest> remoteFirstRequestMap =
+ remoteFirstFetcher.prepareFetchLogRequests();
+ FetchLogRequest remoteFirstRequest = remoteFirstRequestMap.get(1);
+ assertThat(remoteFirstRequest.hasReadPreference()).isTrue();
+ assertThat(remoteFirstRequest.getReadPreference())
+ .isEqualTo(FetchLogReadPreference.REMOTE_FIRST.value());
+ } finally {
+ remoteFirstFetcher.close();
+ }
+ }
+
private LogScannerStatus initializeLogScannerStatus() {
Map<TableBucket, Long> scanBucketAndOffsets = new HashMap<>();
scanBucketAndOffsets.put(tb1, 0L);
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index d9d79502d..233902cb3 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -26,6 +26,7 @@ import org.apache.fluss.metadata.DeleteBehavior;
import org.apache.fluss.metadata.KvFormat;
import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.MergeEngineType;
+import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
import org.apache.fluss.utils.ArrayUtils;
import java.lang.reflect.Field;
@@ -1525,6 +1526,13 @@ public class ConfigOptions {
"The number of remote log segments to keep in local temp file for LogScanner, "
+ "which download from remote storage. The default setting is 4.");
+ public static final ConfigOption<FetchLogReadPreference> CLIENT_SCANNER_LOG_READ_PREFERENCE =
+ key("client.scanner.log.read-preference")
+ .enumType(FetchLogReadPreference.class)
+ .defaultValue(FetchLogReadPreference.LOCAL_FIRST)
+ .withDescription(
+ "The read preference for LogScanner. Supported values are local-first and remote-first.");
+
public static final ConfigOption<String> CLIENT_SCANNER_IO_TMP_DIR =
key("client.scanner.io.tmpdir")
.stringType()
diff --git a/fluss-common/src/main/java/org/apache/fluss/rpc/protocol/FetchLogReadPreference.java b/fluss-common/src/main/java/org/apache/fluss/rpc/protocol/FetchLogReadPreference.java
new file mode 100644
index 000000000..5f70692d9
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/rpc/protocol/FetchLogReadPreference.java
@@ -0,0 +1,52 @@
+/*
+ * 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.rpc.protocol;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/** Read preference for fetch log requests. */
+@PublicEvolving
+public enum FetchLogReadPreference {
+ LOCAL_FIRST(0, "local-first"),
+ REMOTE_FIRST(1, "remote-first");
+ private final int value;
+ private final String name;
+
+ FetchLogReadPreference(int value, String name) {
+ this.value = value;
+ this.name = name;
+ }
+
+ public int value() {
+ return value;
+ }
+
+ public static FetchLogReadPreference from(int value) {
+ for (FetchLogReadPreference preference : values()) {
+ if (preference.value == value) {
+ return preference;
+ }
+ }
+ return LOCAL_FIRST;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java
index 9251e2bfd..16fc0a6db 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java
@@ -45,8 +45,10 @@ import org.apache.flink.streaming.api.graph.StreamGraphHasherV2;
import java.nio.charset.StandardCharsets;
import static org.apache.fluss.config.ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR;
+import static org.apache.fluss.config.ConfigOptions.CLIENT_SCANNER_LOG_READ_PREFERENCE;
import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
import static org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils.getClientScannerIoTmpDir;
+import static org.apache.fluss.rpc.protocol.FetchLogReadPreference.REMOTE_FIRST;
/**
* The flink source implementation for tiering data from Fluss to downstream lake.
@@ -112,10 +114,13 @@ public class TieringSource<WriteResult>
SourceReaderContext sourceReaderContext) {
FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
elementsQueue = new FutureCompletingBlockingQueue<>();
- flussConf.set(
+ Configuration tieringReaderConf = new Configuration(flussConf);
+ tieringReaderConf.set(
CLIENT_SCANNER_IO_TMP_DIR,
- getClientScannerIoTmpDir(flussConf, sourceReaderContext.getConfiguration()));
- Connection connection = ConnectionFactory.createConnection(flussConf);
+ getClientScannerIoTmpDir(
+ tieringReaderConf, sourceReaderContext.getConfiguration()));
+ tieringReaderConf.set(CLIENT_SCANNER_LOG_READ_PREFERENCE, REMOTE_FIRST);
+ Connection connection = ConnectionFactory.createConnection(tieringReaderConf);
return new TieringSourceReader<>(
elementsQueue, sourceReaderContext, connection, lakeTieringFactory);
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java
index aab355718..5cda2d9f5 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java
@@ -173,6 +173,15 @@ class FlinkTieringTestBase {
return TestingValuesLake.getResults(tablePath.toString());
}
+ protected TieringJobScope startTieringJob(StreamExecutionEnvironment execEnv) throws Exception {
+ return new TieringJobScope(buildTieringJob(execEnv));
+ }
+
+ protected TieringJobScope startTieringJob(
+ StreamExecutionEnvironment execEnv, Configuration lakeTieringConfig) throws Exception {
+ return new TieringJobScope(buildTieringJob(execEnv, lakeTieringConfig));
+ }
+
protected JobClient buildTieringJob(StreamExecutionEnvironment execEnv) throws Exception {
return buildTieringJob(execEnv, new Configuration());
}
@@ -190,4 +199,34 @@ class FlinkTieringTestBase {
DataLakeFormat.LANCE.toString())
.build();
}
+
+ /**
+ * A try-with-resources scope that starts a tiering job and cancels it when closed.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * try (TieringJobScope tiering = startTieringJob(execEnv)) {
+ * assertReplicaStatus(tableBucket, expectedLogEndOffset);
+ * }
+ * }</pre>
+ */
+ protected static final class TieringJobScope implements AutoCloseable {
+
+ private final JobClient jobClient;
+
+ private TieringJobScope(JobClient jobClient) {
+ this.jobClient = jobClient;
+ }
+
+ /** Returns the underlying Flink job client. */
+ public JobClient client() {
+ return jobClient;
+ }
+
+ @Override
+ public void close() throws Exception {
+ jobClient.cancel().get();
+ }
+ }
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java
index 86b89e4a7..a8cb354ca 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java
@@ -22,15 +22,18 @@ import org.apache.fluss.client.metadata.LakeSnapshot;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.row.BinaryString;
import org.apache.fluss.row.GenericRow;
import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.server.log.FetchIsolation;
+import org.apache.fluss.server.log.LogTablet;
+import org.apache.fluss.server.replica.Replica;
import org.apache.fluss.types.DataTypes;
import org.apache.fluss.utils.ExceptionUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
-import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
@@ -44,6 +47,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue;
import static org.assertj.core.api.Assertions.assertThat;
@@ -95,22 +99,91 @@ abstract class TieringITCase extends FlinkTieringTestBase {
// set tiering duration to a small value for testing purpose
Configuration lakeTieringConfig = new Configuration();
- JobClient jobClient = buildTieringJob(execEnv, lakeTieringConfig);
-
- try {
- // verify the tiered records is less than the table total record to
- // make sure tiering is forced to complete when reach max duration
- LakeSnapshot logTableLakeSnapshot = waitLakeSnapshot(logTablePath);
- long tieredRecords = countTieredRecords(logTableLakeSnapshot);
- assertThat(tieredRecords).isLessThan(recordCount);
-
- // verify the tiered records is less than the table total record to
- // make sure tiering is forced to complete when reach max duration
- LakeSnapshot pkTableLakeSnapshot = waitLakeSnapshot(pkTablePath);
- tieredRecords = countTieredRecords(pkTableLakeSnapshot);
- assertThat(tieredRecords).isLessThan(recordCount);
- } finally {
- jobClient.cancel();
+ try (TieringJobScope ignored = startTieringJob(execEnv, lakeTieringConfig)) {
+ // Wait until all records are tiered, then verify that max duration forced tiering to
+ // complete in multiple snapshots.
+ LakeSnapshot logTableLakeSnapshot = waitUntilFullyTiered(logTablePath, recordCount);
+ assertThat(countTieredRecords(logTableLakeSnapshot)).isEqualTo(recordCount);
+ assertThat(logTableLakeSnapshot.getSnapshotId()).isGreaterThan(0L);
+
+ LakeSnapshot pkTableLakeSnapshot = waitUntilFullyTiered(pkTablePath, recordCount);
+ assertThat(countTieredRecords(pkTableLakeSnapshot)).isEqualTo(recordCount);
+ assertThat(pkTableLakeSnapshot.getSnapshotId()).isGreaterThan(0L);
+ }
+ }
+
+ @Test
+ void testTieringReadsRemoteFirstAndSwitchesToLocalTail() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "remote_first_log_table");
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.STRING())
+ .build();
+ long tableId = createTable(tablePath, schema);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+
+ int remoteRecordCount = 4;
+ List<InternalRow> expectedRows = createRows(0, remoteRecordCount);
+ writeRows(tablePath, expectedRows, true);
+
+ Replica replica = getLeaderReplica(tableBucket);
+ LogTablet logTablet = replica.getLogTablet();
+ logTablet.roll(Optional.empty());
+
+ FLUSS_CLUSTER_EXTENSION.waitUntilSomeLogSegmentsCopyToRemote(tableBucket);
+ retry(
+ Duration.ofMinutes(1),
+ () -> assertThat(logTablet.canFetchFromRemoteLog(remoteRecordCount - 1L)).isTrue());
+
+ List<InternalRow> localTailRows = createRows(remoteRecordCount, 2);
+ expectedRows.addAll(localTailRows);
+ writeRows(tablePath, localTailRows, true);
+
+ assertThat(logTablet.canFetchFromRemoteLog(remoteRecordCount)).isFalse();
+ assertThat(logTablet.localLogStartOffset()).isZero();
+ assertThat(logTablet.localLogEndOffset()).isEqualTo(expectedRows.size());
+
+ int allLocalBytes = readLocalBytes(logTablet, 0L);
+ int localTailBytes = readLocalBytes(logTablet, remoteRecordCount);
+ assertThat(localTailBytes).isPositive().isLessThan(allLocalBytes);
+
+ long localBytesOutBefore =
+ replica.tableMetrics().getServerMetricGroup().bytesOut().getCount();
+ try (TieringJobScope ignored = startTieringJob(execEnv)) {
+ assertReplicaStatus(tableBucket, expectedRows.size());
+ assertRows(tablePath, expectedRows);
+
+ long localBytesOut =
+ replica.tableMetrics().getServerMetricGroup().bytesOut().getCount()
+ - localBytesOutBefore;
+ assertThat(localBytesOut).isEqualTo(localTailBytes);
+ }
+ }
+
+ private List<InternalRow> createRows(int start, int count) {
+ List<InternalRow> rows = new ArrayList<>();
+ for (int i = start; i < start + count; i++) {
+ rows.add(GenericRow.of(i, BinaryString.fromString("v" + i)));
+ }
+ return rows;
+ }
+
+ private int readLocalBytes(LogTablet logTablet, long offset) throws Exception {
+ return logTablet
+ .read(offset, Integer.MAX_VALUE, FetchIsolation.LOG_END, true, null, null)
+ .getRecords()
+ .sizeInBytes();
+ }
+
+ private void assertRows(TablePath tablePath, List<InternalRow> expectedRows) {
+ List<InternalRow> actualRows = getValuesRecords(tablePath);
+ assertThat(actualRows).hasSameSizeAs(expectedRows);
+ for (int i = 0; i < expectedRows.size(); i++) {
+ InternalRow actual = actualRows.get(i);
+ InternalRow expected = expectedRows.get(i);
+ assertThat(actual.getInt(0)).isEqualTo(expected.getInt(0));
+ assertThat(actual.getString(1)).isEqualTo(expected.getString(1));
}
}
@@ -120,11 +193,14 @@ abstract class TieringITCase extends FlinkTieringTestBase {
.sum();
}
- private LakeSnapshot waitLakeSnapshot(TablePath tablePath) {
+ private LakeSnapshot waitUntilFullyTiered(TablePath tablePath, long expectedRecordCount) {
return waitValue(
() -> {
try {
- return Optional.of(admin.getLatestLakeSnapshot(tablePath).get());
+ LakeSnapshot lakeSnapshot = admin.getLatestLakeSnapshot(tablePath).get();
+ return countTieredRecords(lakeSnapshot) == expectedRecordCount
+ ? Optional.of(lakeSnapshot)
+ : Optional.empty();
} catch (Exception e) {
if (ExceptionUtils.stripExecutionException(e)
instanceof LakeTableSnapshotNotExistException) {
@@ -134,7 +210,7 @@ abstract class TieringITCase extends FlinkTieringTestBase {
}
},
Duration.ofSeconds(30),
- "Fail to wait for one round of tiering finish for table " + tablePath);
+ "Fail to wait for tiering to finish for table " + tablePath);
}
private void createTable(TablePath tablePath, boolean isPrimaryKeyTable) throws Exception {
diff --git a/fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java b/fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
index 0f4d1a610..af08d84d6 100644
--- a/fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
+++ b/fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
@@ -28,7 +28,6 @@ import org.apache.fluss.row.InternalRow;
import org.apache.fluss.types.DataTypes;
import org.apache.flink.api.common.RuntimeExecutionMode;
-import org.apache.flink.core.execution.JobClient;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -97,8 +96,7 @@ class TieringFailoverITCase extends FlinkTieringTestBase {
TestingValuesLake.failWhen(t1.toString()).failWriteNext(2);
// then start tiering job
- JobClient jobClient = buildTieringJob(execEnv);
- try {
+ try (TieringJobScope ignored = startTieringJob(execEnv)) {
// check the status of replica after synced
assertReplicaStatus(t1Bucket, 3);
@@ -117,10 +115,6 @@ class TieringFailoverITCase extends FlinkTieringTestBase {
assertReplicaStatus(t1Bucket, expectedRows.size());
checkDataInValuesTable(t1, expectedRows);
- } finally {
- if (!jobClient.getJobExecutionResult().isDone()) {
- jobClient.cancel().get();
- }
}
}
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 57e96092d..836c642a7 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -229,6 +229,7 @@ message FetchLogRequest {
repeated PbFetchLogReqForTable tables_req = 3;
optional int32 max_wait_ms = 4;
optional int32 min_bytes = 5;
+ optional int32 read_preference = 6;
}
message FetchLogResponse {
@@ -1410,4 +1411,4 @@ message PbLiteralValue {
optional int64 timestamp_millis_value = 11; // Epoch millis
optional int32 timestamp_nano_of_millis_value = 12; // Nano of millis
optional bytes decimal_bytes = 13; // Serialized decimal (non-compact mode)
-}
\ No newline at end of file
+}
diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs
index aadd1cb86..5750a807b 100644
--- a/fluss-rust/crates/fluss/src/client/table/scanner.rs
+++ b/fluss-rust/crates/fluss/src/client/table/scanner.rs
@@ -2185,6 +2185,7 @@ impl LogFetcher {
tables_req: vec![req_for_table],
max_wait_ms: Some(self.fetch_wait_max_time_ms),
min_bytes: Some(self.fetch_min_bytes),
+ read_preference: None,
};
(leader_id, fetch_log_request)
})
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
index 084529de2..76a30a1dd 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
@@ -23,6 +23,7 @@ import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.record.FileLogProjection;
import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.rpc.messages.FetchLogRequest;
+import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
import javax.annotation.Nullable;
@@ -69,14 +70,29 @@ public final class FetchParams {
@Nullable private FileLogProjection fileLogProjection;
private final int minFetchBytes;
private final long maxWaitMs;
+ private final FetchLogReadPreference readPreference;
// TODO: add more params like epoch etc.
public FetchParams(int replicaId, int maxFetchBytes) {
- this(replicaId, true, maxFetchBytes, DEFAULT_MIN_FETCH_BYTES, DEFAULT_MAX_WAIT_MS, null);
+ this(
+ replicaId,
+ true,
+ maxFetchBytes,
+ DEFAULT_MIN_FETCH_BYTES,
+ DEFAULT_MAX_WAIT_MS,
+ null,
+ FetchLogReadPreference.LOCAL_FIRST);
}
public FetchParams(int replicaId, int maxFetchBytes, int minFetchBytes, long maxWaitMs) {
- this(replicaId, true, maxFetchBytes, minFetchBytes, maxWaitMs, null);
+ this(
+ replicaId,
+ true,
+ maxFetchBytes,
+ minFetchBytes,
+ maxWaitMs,
+ null,
+ FetchLogReadPreference.LOCAL_FIRST);
}
@VisibleForTesting
@@ -86,7 +102,8 @@ public final class FetchParams {
int maxFetchBytes,
int minFetchBytes,
long maxWaitMs,
- @Nullable Map<Long, FilterInfo> tableFilterInfoMap) {
+ @Nullable Map<Long, FilterInfo> tableFilterInfoMap,
+ FetchLogReadPreference readPreference) {
this.replicaId = replicaId;
this.fetchOnlyLeader = fetchOnlyLeader;
this.maxFetchBytes = maxFetchBytes;
@@ -96,6 +113,25 @@ public final class FetchParams {
this.minFetchBytes = minFetchBytes;
this.maxWaitMs = maxWaitMs;
this.tableFilterInfoMap = tableFilterInfoMap;
+ this.readPreference = readPreference;
+ }
+
+ @VisibleForTesting
+ public FetchParams(
+ int replicaId,
+ boolean fetchOnlyLeader,
+ int maxFetchBytes,
+ int minFetchBytes,
+ long maxWaitMs,
+ @Nullable Map<Long, FilterInfo> tableFilterInfoMap) {
+ this(
+ replicaId,
+ fetchOnlyLeader,
+ maxFetchBytes,
+ minFetchBytes,
+ maxWaitMs,
+ tableFilterInfoMap,
+ FetchLogReadPreference.LOCAL_FIRST);
}
public void setCurrentFetch(
@@ -183,6 +219,14 @@ public final class FetchParams {
return replicaId >= 0;
}
+ public FetchLogReadPreference readPreference() {
+ return readPreference;
+ }
+
+ public boolean isRemoteFirstClientFetch() {
+ return !isFromFollower() && readPreference == FetchLogReadPreference.REMOTE_FIRST;
+ }
+
public boolean fetchOnlyLeader() {
return isFromFollower() || fetchOnlyLeader;
}
@@ -204,12 +248,19 @@ public final class FetchParams {
&& maxFetchBytes == that.maxFetchBytes
&& minFetchBytes == that.minFetchBytes
&& maxWaitMs == that.maxWaitMs
+ && readPreference == that.readPreference
&& Objects.equals(tableFilterInfoMap, that.tableFilterInfoMap);
}
@Override
public int hashCode() {
- return Objects.hash(replicaId, maxFetchBytes, minFetchBytes, maxWaitMs, tableFilterInfoMap);
+ return Objects.hash(
+ replicaId,
+ maxFetchBytes,
+ minFetchBytes,
+ maxWaitMs,
+ tableFilterInfoMap,
+ readPreference);
}
@Override
@@ -225,6 +276,8 @@ public final class FetchParams {
+ maxWaitMs
+ ", tableFilterInfoMap="
+ tableFilterInfoMap
+ + ", readPreference="
+ + readPreference
+ ')';
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParamsBuilder.java b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParamsBuilder.java
index 0bb2c0058..b8de03cf8 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParamsBuilder.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParamsBuilder.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server.log;
import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
import javax.annotation.Nullable;
@@ -34,6 +35,7 @@ public final class FetchParamsBuilder {
@Nullable private Map<Long, FilterInfo> tableFilterInfoMap;
private int minFetchBytes;
private long maxWaitMs;
+ private FetchLogReadPreference readPreference = FetchLogReadPreference.LOCAL_FIRST;
public FetchParamsBuilder(int replicaId, int maxFetchBytes) {
this.replicaId = replicaId;
@@ -63,6 +65,11 @@ public final class FetchParamsBuilder {
return this;
}
+ public FetchParamsBuilder withReadPreference(FetchLogReadPreference readPreference) {
+ this.readPreference = readPreference;
+ return this;
+ }
+
public FetchParams build() {
return new FetchParams(
replicaId,
@@ -70,6 +77,7 @@ public final class FetchParamsBuilder {
maxFetchBytes,
minFetchBytes,
maxWaitMs,
- tableFilterInfoMap);
+ tableFilterInfoMap,
+ readPreference);
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index e38537e27..88d5840e5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -1474,6 +1474,22 @@ public class ReplicaManager implements ServerReconfigurable {
replica.getArrowCompressionInfo(),
fetchReqInfo.getProjectFields(),
projectionsCache);
+
+ // If the client prefers remote reads and the offset is covered, return remote fetch
+ // info.
+ if (fetchParams.isRemoteFirstClientFetch()) {
+ FetchLogResultForBucket remoteFirstFetchResult =
+ tryFetchRemoteFirst(replica, fetchOffset);
+ if (remoteFirstFetchResult != null) {
+ logReadResult.put(
+ tb,
+ new LogReadResult(
+ remoteFirstFetchResult,
+ LogOffsetMetadata.UNKNOWN_OFFSET_METADATA));
+ continue;
+ }
+ }
+
LogReadInfo readInfo = replica.fetchRecords(fetchParams);
// Once we read from a non-empty bucket, we stop ignoring request and bucket
@@ -1532,6 +1548,46 @@ public class ReplicaManager implements ServerReconfigurable {
return logReadResult;
}
+ private @Nullable FetchLogResultForBucket tryFetchRemoteFirst(
+ Replica replica, long fetchOffset) {
+ if (!replica.isLeader()) {
+ throw new NotLeaderOrFollowerException(
+ String.format(
+ "Leader not local for bucket %s on tabletServer %d",
+ replica.getTableBucket(), serverId));
+ }
+
+ TableBucket tb = replica.getTableBucket();
+ long normalizedFetchOffset =
+ fetchOffset == FetchParams.FETCH_FROM_EARLIEST_OFFSET
+ ? replica.getLogStartOffset()
+ : fetchOffset;
+ if (!canFetchFromRemoteLog(replica, normalizedFetchOffset)) {
+ return null;
+ }
+
+ try {
+ RemoteLogFetchInfo remoteLogFetchInfo =
+ fetchLogFromRemote(replica, normalizedFetchOffset);
+ if (remoteLogFetchInfo != null) {
+ return new FetchLogResultForBucket(
+ tb, remoteLogFetchInfo, replica.getLogHighWatermark());
+ }
+ // Remote log is expected to cover the offset, but segments/manifest may not be ready
+ // yet.
+ // REMOTE_FIRST is a preference, so fall back to local reads.
+ return null;
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to fetch remote log first for replica {} at offset {}; falling back to local reads",
+ tb,
+ normalizedFetchOffset,
+ e);
+ // Fall back to local reads on remote fetch failures.
+ return null;
+ }
+ }
+
private FetchLogResultForBucket handleFetchOutOfRangeException(
Replica replica, long fetchOffset, Exception e) {
TableBucket tb = replica.getTableBucket();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
index a18b6289d..0c2058915 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
@@ -76,6 +76,7 @@ import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
import org.apache.fluss.rpc.messages.UpdateMetadataResponse;
import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.rpc.protocol.Errors;
+import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
import org.apache.fluss.rpc.protocol.MergeMode;
import org.apache.fluss.security.acl.OperationType;
import org.apache.fluss.security.acl.Resource;
@@ -248,6 +249,10 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
private static FetchParams getFetchParams(FetchLogRequest request) {
FetchParams fetchParams;
Map<Long, FilterInfo> tableFilterInfoMap = getTableFilterInfoMap(request);
+ FetchLogReadPreference readPreference =
+ request.hasReadPreference()
+ ? FetchLogReadPreference.from(request.getReadPreference())
+ : FetchLogReadPreference.LOCAL_FIRST;
if (request.hasMinBytes()) {
fetchParams =
new FetchParamsBuilder(request.getFollowerServerId(), request.getMaxBytes())
@@ -257,11 +262,13 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
? request.getMaxWaitMs()
: DEFAULT_MAX_WAIT_MS_WHEN_MIN_BYTES_ENABLE)
.withTableFilterInfoMap(tableFilterInfoMap)
+ .withReadPreference(readPreference)
.build();
} else {
fetchParams =
new FetchParamsBuilder(request.getFollowerServerId(), request.getMaxBytes())
.withTableFilterInfoMap(tableFilterInfoMap)
+ .withReadPreference(readPreference)
.build();
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java
index 2944f1c66..96e8ae704 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java
@@ -25,8 +25,10 @@ import org.apache.fluss.remote.RemoteLogFetchInfo;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.protocol.ApiError;
+import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
import org.apache.fluss.server.coordinator.TestCoordinatorGateway;
import org.apache.fluss.server.entity.FetchReqInfo;
+import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.entity.StopReplicaData;
import org.apache.fluss.server.entity.StopReplicaResultForBucket;
import org.apache.fluss.server.log.FetchParams;
@@ -34,6 +36,7 @@ import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.replica.Replica;
import org.apache.fluss.server.replica.ReplicaManager;
import org.apache.fluss.server.testutils.ServerTestTags;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
@@ -59,6 +62,8 @@ import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH_PK;
+import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH;
+import static org.apache.fluss.server.zk.data.LeaderAndIsr.INITIAL_BUCKET_EPOCH;
import static org.apache.fluss.server.zk.data.LeaderAndIsr.INITIAL_LEADER_EPOCH;
import static org.apache.fluss.utils.FlussPaths.remoteLogDir;
import static org.apache.fluss.utils.FlussPaths.remoteLogTabletDir;
@@ -366,6 +371,100 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
assertThat(resultForBucket.fetchFromRemote()).isFalse();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testRemoteFirstFetchPrefersRemoteWhenLocalStillHasRecords(boolean partitionTable)
+ throws Exception {
+ TableBucket tb = makeTableBucket(partitionTable);
+ makeLogTableAsLeader(tb, partitionTable);
+ LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+ logTablet.updateRemoteLogEndOffset(40L);
+
+ Map<TableBucket, FetchReqInfo> fetchData =
+ Collections.singletonMap(tb, new FetchReqInfo(tb.getTableId(), 35L, 1024 * 1024));
+
+ CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> localFirstFuture =
+ new CompletableFuture<>();
+ replicaManager.fetchLogRecords(
+ new FetchParams(-1, Integer.MAX_VALUE),
+ fetchData,
+ null,
+ localFirstFuture::complete);
+ FetchLogResultForBucket localFirstResult = localFirstFuture.get().get(tb);
+ assertThat(localFirstResult.getError()).isEqualTo(ApiError.NONE);
+ assertThat(localFirstResult.fetchFromRemote()).isFalse();
+ assertThat(localFirstResult.records()).isNotNull();
+
+ CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> remoteFirstFuture =
+ new CompletableFuture<>();
+ replicaManager.fetchLogRecords(
+ new FetchParams(
+ -1,
+ true,
+ Integer.MAX_VALUE,
+ -1,
+ -1,
+ null,
+ FetchLogReadPreference.REMOTE_FIRST),
+ fetchData,
+ null,
+ remoteFirstFuture::complete);
+ FetchLogResultForBucket remoteFirstResult = remoteFirstFuture.get().get(tb);
+ assertThat(remoteFirstResult.getError()).isEqualTo(ApiError.NONE);
+ assertThat(remoteFirstResult.fetchFromRemote()).isTrue();
+ assertThat(remoteFirstResult.records()).isNull();
+ assertThat(remoteFirstResult.remoteLogFetchInfo()).isNotNull();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testRemoteFirstFetchRejectsNonLeader(boolean partitionTable) throws Exception {
+ TableBucket tb = makeTableBucket(partitionTable);
+ makeLogTableAsLeader(tb, partitionTable);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+ logTablet.updateRemoteLogEndOffset(40L);
+
+ int newLeaderId = TABLET_SERVER_ID + 1;
+ replica.makeFollower(
+ new NotifyLeaderAndIsrData(
+ partitionTable
+ ? DATA1_PHYSICAL_TABLE_PATH_PA_2024
+ : DATA1_PHYSICAL_TABLE_PATH,
+ tb,
+ Arrays.asList(TABLET_SERVER_ID, newLeaderId),
+ new LeaderAndIsr(
+ newLeaderId,
+ INITIAL_LEADER_EPOCH + 1,
+ Arrays.asList(TABLET_SERVER_ID, newLeaderId),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH + 1)));
+
+ CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> fetchFuture =
+ new CompletableFuture<>();
+ replicaManager.fetchLogRecords(
+ new FetchParams(
+ -1,
+ true,
+ Integer.MAX_VALUE,
+ -1,
+ -1,
+ null,
+ FetchLogReadPreference.REMOTE_FIRST),
+ Collections.singletonMap(tb, new FetchReqInfo(tb.getTableId(), 35L, 1024 * 1024)),
+ null,
+ fetchFuture::complete);
+
+ FetchLogResultForBucket result = fetchFuture.get().get(tb);
+ assertThat(result.getError().exception()).isInstanceOf(NotLeaderOrFollowerException.class);
+ assertThat(result.fetchFromRemote()).isFalse();
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testCleanupLocalSegments(boolean partitionTable) throws Exception {