This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c5884b622a [flink] Add partial lookup metrics to query service (#8822)
c5884b622a is described below
commit c5884b622a6cbddc7c2607f3e3ad3539b3b6f4e9
Author: sanshi <[email protected]>
AuthorDate: Thu Jul 23 21:13:56 2026 +0800
[flink] Add partial lookup metrics to query service (#8822)
---
.../flink/service/QueryExecutorOperator.java | 12 +++++-
.../paimon/flink/RemoteLookupJoinITCase.java | 45 ++++++++++++++++++++--
2 files changed, 52 insertions(+), 5 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
index bf0521d550..cd81dfa35a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
@@ -23,9 +23,11 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
import org.apache.paimon.flink.utils.RuntimeContextUtils;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.operation.metrics.PartialLookupMetrics;
import org.apache.paimon.service.network.NetworkUtils;
import org.apache.paimon.service.network.stats.DisabledServiceRequestStats;
import org.apache.paimon.service.server.KvQueryServer;
@@ -75,7 +77,15 @@ public class QueryExecutorOperator extends
AbstractStreamOperator<InternalRow>
.getEnvironment()
.getIOManager()
.getSpillingDirectoriesPaths());
- this.query = ((FileStoreTable)
table).newLocalTableQuery().withIOManager(ioManager);
+ PartialLookupMetrics metrics =
+ new PartialLookupMetrics(
+ new
FlinkMetricRegistry(getRuntimeContext().getMetricGroup()),
+ table.name());
+ this.query =
+ ((FileStoreTable) table)
+ .newLocalTableQuery()
+ .withIOManager(ioManager)
+ .withMetrics(metrics);
KvQueryServer server =
new KvQueryServer(
RuntimeContextUtils.getIndexOfThisSubtask(getRuntimeContext()),
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
index fca350da19..0854272e46 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
@@ -24,6 +24,8 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.flink.query.RemoteTableQuery;
import org.apache.paimon.flink.service.QueryService;
+import org.apache.paimon.metrics.MetricGroupImpl;
+import org.apache.paimon.operation.metrics.PartialLookupMetrics;
import org.apache.paimon.service.ServiceManager;
import org.apache.paimon.service.network.stats.DisabledServiceRequestStats;
import org.apache.paimon.service.server.KvQueryServer;
@@ -143,6 +145,25 @@ public class RemoteLookupJoinITCase extends
CatalogITCaseBase {
proxy.close();
}
+ @Test
+ public void testRemoteQueryServicePartialLookupMetrics() throws Throwable {
+ sql("CREATE TABLE DIM (k INT PRIMARY KEY NOT ENFORCED, v INT) WITH
('bucket' = '1')");
+ ServiceProxy proxy = launchQueryServer("DIM");
+ proxy.write(GenericRow.of(1, 11));
+
+ RemoteTableQuery query = new RemoteTableQuery(paimonTable("DIM"));
+ assertThat(query.lookup(row(), 0, row(1))).isNotNull();
+ assertThat(proxy.metrics().lookupCount()).isEqualTo(1);
+ assertThat(proxy.metrics().remoteAccessCount()).isEqualTo(1);
+
+ assertThat(query.lookup(row(), 0, row(1))).isNotNull();
+ assertThat(proxy.metrics().lookupCount()).isEqualTo(2);
+ assertThat(proxy.metrics().remoteAccessCount()).isEqualTo(1);
+
+ query.close();
+ proxy.close();
+ }
+
@Disabled // TODO unstable
@Test
public void testServiceFileCleaned() throws Exception {
@@ -211,7 +232,14 @@ public class RemoteLookupJoinITCase extends
CatalogITCaseBase {
private ServiceProxy launchQueryServer(String tableName) throws Throwable {
FileStoreTable table = (FileStoreTable) paimonTable(tableName);
- LocalTableQuery query =
table.newLocalTableQuery().withIOManager(IOManager.create(path));
+ PartialLookupMetrics metrics =
+ new PartialLookupMetrics(
+ (groupName, variables) -> new
MetricGroupImpl(groupName, variables),
+ table.name());
+ LocalTableQuery query =
+ table.newLocalTableQuery()
+ .withIOManager(IOManager.create(path))
+ .withMetrics(metrics);
KvQueryServer server =
new KvQueryServer(
0,
@@ -231,11 +259,13 @@ public class RemoteLookupJoinITCase extends
CatalogITCaseBase {
return new ServiceProxy() {
@Override
- public void write(InternalRow row) throws Exception {
+ public void write(InternalRow... rows) throws Exception {
BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
BatchTableWrite write = writeBuilder.newWrite();
BatchTableCommit commit = writeBuilder.newCommit();
- write.write(row);
+ for (InternalRow row : rows) {
+ write.write(row);
+ }
List<CommitMessage> commitMessages = write.prepareCommit();
commit.commit(commitMessages);
@@ -247,6 +277,11 @@ public class RemoteLookupJoinITCase extends
CatalogITCaseBase {
message.newFilesIncrement().newFiles());
}
+ @Override
+ public PartialLookupMetrics metrics() {
+ return metrics;
+ }
+
@Override
public void close() throws IOException {
server.shutdown();
@@ -257,6 +292,8 @@ public class RemoteLookupJoinITCase extends
CatalogITCaseBase {
private interface ServiceProxy extends Closeable {
- void write(InternalRow row) throws Exception;
+ void write(InternalRow... rows) throws Exception;
+
+ PartialLookupMetrics metrics();
}
}
