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 09fc333a20 [flink] Add metrics for local partial lookup remote access
(#8717)
09fc333a20 is described below
commit 09fc333a20f2303b9210dd2f287c021601717a2a
Author: sanshi <[email protected]>
AuthorDate: Mon Jul 20 12:03:35 2026 +0800
[flink] Add metrics for local partial lookup remote access (#8717)
---
.../org/apache/paimon/mergetree/LookupLevels.java | 52 +++++++++++++---
.../operation/metrics/PartialLookupMetrics.java | 69 ++++++++++++++++++++++
.../apache/paimon/table/query/LocalTableQuery.java | 29 ++++++++-
.../metrics/PartialLookupMetricsTest.java | 56 ++++++++++++++++++
.../paimon/table/PrimaryKeySimpleTableTest.java | 9 ++-
.../flink/lookup/FileStoreLookupFunction.java | 24 +++++++-
.../flink/lookup/PrimaryKeyPartialLookupTable.java | 21 ++++++-
7 files changed, 246 insertions(+), 14 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
index 850892a8c0..0687323e60 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
@@ -135,23 +135,40 @@ public class LookupLevels<T> implements
Levels.DropFileCallback, Closeable {
@Nullable
public T lookup(InternalRow key, int startLevel) throws IOException {
- return LookupUtils.lookup(levels, key, startLevel, this::lookup,
this::lookupLevel0);
+ return lookup(key, startLevel, null);
}
@Nullable
- private T lookupLevel0(InternalRow key, TreeSet<DataFileMeta> level0)
throws IOException {
- return LookupUtils.lookupLevel0(keyComparator, key, level0,
this::lookup);
+ public T lookup(InternalRow key, int startLevel, @Nullable LookupContext
context)
+ throws IOException {
+ return LookupUtils.lookup(
+ levels,
+ key,
+ startLevel,
+ (lookupKey, level) -> lookup(lookupKey, level, context),
+ (lookupKey, level0) -> lookupLevel0(lookupKey, level0,
context));
+ }
+
+ @Nullable
+ private T lookupLevel0(
+ InternalRow key, TreeSet<DataFileMeta> level0, @Nullable
LookupContext context)
+ throws IOException {
+ return LookupUtils.lookupLevel0(
+ keyComparator, key, level0, (lookupKey, file) ->
lookup(lookupKey, file, context));
}
@Nullable
- private T lookup(InternalRow key, SortedRun level) throws IOException {
- return LookupUtils.lookup(keyComparator, key, level, this::lookup);
+ private T lookup(InternalRow key, SortedRun level, @Nullable LookupContext
context)
+ throws IOException {
+ return LookupUtils.lookup(
+ keyComparator, key, level, (lookupKey, file) ->
lookup(lookupKey, file, context));
}
@Nullable
- private T lookup(InternalRow key, DataFileMeta file) throws IOException {
+ private T lookup(InternalRow key, DataFileMeta file, @Nullable
LookupContext context)
+ throws IOException {
byte[] keyBytes = serializeKey(key);
- LookupResult lookupResult = lookupFile(file, keyBytes);
+ LookupResult lookupResult = lookupFile(file, keyBytes, context);
byte[] valueBytes = lookupResult.valueBytes;
if (valueBytes == null) {
return null;
@@ -165,7 +182,9 @@ public class LookupLevels<T> implements
Levels.DropFileCallback, Closeable {
file.fileName());
}
- private LookupResult lookupFile(DataFileMeta file, byte[] keyBytes) throws
IOException {
+ private LookupResult lookupFile(
+ DataFileMeta file, byte[] keyBytes, @Nullable LookupContext
context)
+ throws IOException {
String fileName = file.fileName();
LookupFile lookupFile = lookupFileCache.getIfPresent(fileName);
LookupResult lookupResult = lookupCachedFile(fileName, lookupFile,
keyBytes);
@@ -181,6 +200,9 @@ public class LookupLevels<T> implements
Levels.DropFileCallback, Closeable {
return lookupResult;
}
+ if (context != null) {
+ context.markRemoteAccessed();
+ }
lookupFile = createLookupFile(file);
try {
@@ -191,6 +213,20 @@ public class LookupLevels<T> implements
Levels.DropFileCallback, Closeable {
}
}
+ /** Tracks whether one lookup invocation created any lookup file from
table storage. */
+ public static class LookupContext {
+
+ private boolean remoteAccessed;
+
+ private void markRemoteAccessed() {
+ remoteAccessed = true;
+ }
+
+ public boolean remoteAccessed() {
+ return remoteAccessed;
+ }
+ }
+
private Object lookupFileLock(String fileName) {
return lookupFileLocks[Math.floorMod(fileName.hashCode(),
lookupFileLocks.length)];
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/PartialLookupMetrics.java
b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/PartialLookupMetrics.java
new file mode 100644
index 0000000000..23aab2ec79
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/PartialLookupMetrics.java
@@ -0,0 +1,69 @@
+/*
+ * 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.paimon.operation.metrics;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.metrics.Counter;
+import org.apache.paimon.metrics.MetricGroup;
+import org.apache.paimon.metrics.MetricRegistry;
+
+/**
+ * Request-level metrics for local partial lookup. A remote access means that
at least one lookup
+ * file had to be created from table storage during the request.
+ */
+public class PartialLookupMetrics {
+
+ public static final String GROUP_NAME = "lookup";
+ public static final String PARTIAL_LOOKUP_COUNT = "partialLookupCount";
+ public static final String PARTIAL_LOOKUP_REMOTE_ACCESS_COUNT =
+ "partialLookupRemoteAccessCount";
+
+ private final MetricGroup metricGroup;
+ private final Counter lookupCount;
+ private final Counter remoteAccessCount;
+
+ public PartialLookupMetrics(MetricRegistry registry, String tableName) {
+ this.metricGroup = registry.createTableMetricGroup(GROUP_NAME,
tableName);
+ this.lookupCount = metricGroup.counter(PARTIAL_LOOKUP_COUNT);
+ this.remoteAccessCount =
metricGroup.counter(PARTIAL_LOOKUP_REMOTE_ACCESS_COUNT);
+ }
+
+ /** Reports one lookup invocation and whether it accessed table storage. */
+ public void reportLookup(boolean remoteAccessed) {
+ lookupCount.inc();
+ if (remoteAccessed) {
+ remoteAccessCount.inc();
+ }
+ }
+
+ @VisibleForTesting
+ public MetricGroup metricGroup() {
+ return metricGroup;
+ }
+
+ @VisibleForTesting
+ public long lookupCount() {
+ return lookupCount.getCount();
+ }
+
+ @VisibleForTesting
+ public long remoteAccessCount() {
+ return remoteAccessCount.getCount();
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java
b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java
index fdc57861b6..b42e16f5af 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java
@@ -40,6 +40,7 @@ import org.apache.paimon.mergetree.LookupLevels;
import org.apache.paimon.mergetree.lookup.LookupSerializerFactory;
import org.apache.paimon.mergetree.lookup.PersistValueProcessor;
import org.apache.paimon.mergetree.lookup.RemoteLookupFileManager;
+import org.apache.paimon.operation.metrics.PartialLookupMetrics;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
@@ -89,6 +90,8 @@ public class LocalTableQuery implements TableQuery {
@Nullable private Filter<InternalRow> cacheRowFilter;
+ @Nullable private PartialLookupMetrics partialLookupMetrics;
+
public LocalTableQuery(FileStoreTable table) {
this.options = table.coreOptions();
this.tableView = new ConcurrentHashMap<>();
@@ -221,6 +224,25 @@ public class LocalTableQuery implements TableQuery {
@Nullable
@Override
public InternalRow lookup(BinaryRow partition, int bucket, InternalRow
key) throws IOException {
+ PartialLookupMetrics currentMetrics = partialLookupMetrics;
+ LookupLevels.LookupContext context =
+ currentMetrics == null ? null : new
LookupLevels.LookupContext();
+ try {
+ return lookup(partition, bucket, key, context);
+ } finally {
+ if (currentMetrics != null) {
+ currentMetrics.reportLookup(context != null &&
context.remoteAccessed());
+ }
+ }
+ }
+
+ @Nullable
+ private InternalRow lookup(
+ BinaryRow partition,
+ int bucket,
+ InternalRow key,
+ @Nullable LookupLevels.LookupContext context)
+ throws IOException {
Map<Integer, BucketLookupState> buckets = tableView.get(partition);
if (buckets == null || buckets.isEmpty()) {
return null;
@@ -237,7 +259,7 @@ public class LocalTableQuery implements TableQuery {
return null;
}
- KeyValue kv = lookupLevels.lookup(key, startLevel);
+ KeyValue kv = lookupLevels.lookup(key, startLevel, context);
if (kv == null || kv.valueKind().isRetract()) {
return null;
} else {
@@ -264,6 +286,11 @@ public class LocalTableQuery implements TableQuery {
return this;
}
+ public LocalTableQuery withMetrics(@Nullable PartialLookupMetrics metrics)
{
+ this.partialLookupMetrics = metrics;
+ return this;
+ }
+
@Override
public InternalRowSerializer createValueSerializer() {
return
InternalSerializers.create(readerFactoryBuilder.readValueType());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/metrics/PartialLookupMetricsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/PartialLookupMetricsTest.java
new file mode 100644
index 0000000000..1cfa34cfb1
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/PartialLookupMetricsTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.paimon.operation.metrics;
+
+import org.apache.paimon.metrics.Metric;
+import org.apache.paimon.metrics.MetricGroup;
+import org.apache.paimon.metrics.TestMetricRegistry;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link PartialLookupMetrics}. */
+public class PartialLookupMetricsTest {
+
+ @Test
+ public void testRegistrationAndReporting() {
+ PartialLookupMetrics metrics =
+ new PartialLookupMetrics(new TestMetricRegistry(), "myTable");
+ MetricGroup metricGroup = metrics.metricGroup();
+
+
assertThat(metricGroup.getGroupName()).isEqualTo(PartialLookupMetrics.GROUP_NAME);
+ assertThat(metricGroup.getAllVariables()).containsEntry("table",
"myTable");
+ Map<String, Metric> registeredMetrics = metricGroup.getMetrics();
+ assertThat(registeredMetrics.keySet())
+ .containsExactlyInAnyOrder(
+ PartialLookupMetrics.PARTIAL_LOOKUP_COUNT,
+
PartialLookupMetrics.PARTIAL_LOOKUP_REMOTE_ACCESS_COUNT);
+
+ metrics.reportLookup(false);
+ assertThat(metrics.lookupCount()).isEqualTo(1);
+ assertThat(metrics.remoteAccessCount()).isZero();
+
+ metrics.reportLookup(true);
+ assertThat(metrics.lookupCount()).isEqualTo(2);
+ assertThat(metrics.remoteAccessCount()).isEqualTo(1);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index dd3a21b86c..ea8f096a53 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -36,8 +36,10 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.metrics.TestMetricRegistry;
import org.apache.paimon.operation.AbstractFileStoreWrite;
import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.operation.metrics.PartialLookupMetrics;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.postpone.PostponeBucketFileStoreWrite;
@@ -2381,7 +2383,10 @@ public class PrimaryKeySimpleTableTest extends
SimpleTableTestBase {
}
// full value (no projection) -> downloader is wired -> lookup
succeeds via download
- LocalTableQuery query =
table.newLocalTableQuery().withIOManager(ioManager);
+ PartialLookupMetrics metrics =
+ new PartialLookupMetrics(new TestMetricRegistry(),
table.name());
+ LocalTableQuery query =
+
table.newLocalTableQuery().withIOManager(ioManager).withMetrics(metrics);
for (DataSplit split : dataSplits) {
query.refreshFiles(
split.partition(), split.bucket(),
Collections.emptyList(), split.dataFiles());
@@ -2394,6 +2399,8 @@ public class PrimaryKeySimpleTableTest extends
SimpleTableTestBase {
assertThat(value).isNotNull();
assertThat(BATCH_ROW_TO_STRING.apply(value))
.isEqualTo("1|20|200|binary|varbinary|mapKey:mapVal|multiset");
+ assertThat(metrics.lookupCount()).isEqualTo(2);
+ assertThat(metrics.remoteAccessCount()).isEqualTo(1);
query.close();
// value projection -> remote sst (full value) is unsafe to reuse, so
the downloader is
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
index 77d74526a6..f7c6fb2d67 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -28,8 +28,11 @@ import org.apache.paimon.flink.FlinkRowData;
import org.apache.paimon.flink.FlinkRowDataWithBlob;
import org.apache.paimon.flink.FlinkRowWrapper;
import org.apache.paimon.flink.lookup.partitioner.ShuffleStrategy;
+import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
import org.apache.paimon.flink.utils.RuntimeContextUtils;
import org.apache.paimon.flink.utils.TableScanUtils;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.operation.metrics.PartialLookupMetrics;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.ChainGroupReadTable;
@@ -103,6 +106,8 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
private transient File path;
private transient String tmpDirectory;
private transient LookupTable lookupTable;
+ @Nullable private transient MetricRegistry metricRegistry;
+ @Nullable private transient PartialLookupMetrics partialLookupMetrics;
// partition refresh
@Nullable private transient PartitionRefresher partitionRefresher;
@@ -179,6 +184,7 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
public void open(FunctionContext context) throws Exception {
this.functionContext = context;
+ this.metricRegistry = new
FlinkMetricRegistry(context.getMetricGroup());
this.tmpDirectory = getTmpDirectory(context);
open(tmpDirectory);
}
@@ -225,7 +231,12 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
try {
this.lookupTable =
PrimaryKeyPartialLookupTable.createLocalTable(
- table, projection, path, joinKeys,
getRequireCachedBucketIds());
+ table,
+ projection,
+ path,
+ joinKeys,
+ getRequireCachedBucketIds(),
+ this::partialLookupMetrics);
LOG.info(
"Remote service isn't available. Created
PrimaryKeyPartialLookupTable with LocalQueryExecutor.");
} catch (UnsupportedOperationException e) {
@@ -277,6 +288,17 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
lookupTable.open();
}
+ @Nullable
+ private PartialLookupMetrics partialLookupMetrics() {
+ if (metricRegistry == null) {
+ return null;
+ }
+ if (partialLookupMetrics == null) {
+ partialLookupMetrics = new PartialLookupMetrics(metricRegistry,
table.name());
+ }
+ return partialLookupMetrics;
+ }
+
@Nullable
private Predicate createProjectedPredicate(int[] projection) {
Predicate adjustedPredicate = null;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
index 9964a8fa6c..c88112844f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
@@ -28,6 +28,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.flink.query.RemoteTableQuery;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.operation.metrics.PartialLookupMetrics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.BucketMode;
@@ -53,6 +54,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.Supplier;
import static java.util.Collections.emptyList;
import static org.apache.paimon.table.BucketMode.POSTPONE_BUCKET;
@@ -217,6 +219,16 @@ public class PrimaryKeyPartialLookupTable implements
LookupTable {
File tempPath,
List<String> joinKey,
Set<Integer> requireCachedBucketIds) {
+ return createLocalTable(table, projection, tempPath, joinKey,
requireCachedBucketIds, null);
+ }
+
+ public static PrimaryKeyPartialLookupTable createLocalTable(
+ FileStoreTable table,
+ int[] projection,
+ File tempPath,
+ List<String> joinKey,
+ Set<Integer> requireCachedBucketIds,
+ @Nullable Supplier<PartialLookupMetrics> metricsSupplier) {
return new PrimaryKeyPartialLookupTable(
(filter, cacheRowFilter) ->
new LocalQueryExecutor(
@@ -225,7 +237,8 @@ public class PrimaryKeyPartialLookupTable implements
LookupTable {
tempPath,
filter,
requireCachedBucketIds,
- cacheRowFilter),
+ cacheRowFilter,
+ metricsSupplier == null ? null :
metricsSupplier.get()),
table,
joinKey);
}
@@ -273,11 +286,13 @@ public class PrimaryKeyPartialLookupTable implements
LookupTable {
File tempPath,
@Nullable Predicate filter,
Set<Integer> requireCachedBucketIds,
- @Nullable Filter<InternalRow> cacheRowFilter) {
+ @Nullable Filter<InternalRow> cacheRowFilter,
+ @Nullable PartialLookupMetrics metrics) {
this.tableQuery =
table.newLocalTableQuery()
.withValueProjection(projection)
- .withIOManager(new
IOManagerImpl(tempPath.toString()));
+ .withIOManager(new
IOManagerImpl(tempPath.toString()))
+ .withMetrics(metrics);
if (cacheRowFilter != null) {
this.tableQuery.withCacheRowFilter(cacheRowFilter);
