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 38fae3467 [client] Avoid per-record AutoPartitionStrategy construction
in WriterClient.doSend in log tables (#3529)
38fae3467 is described below
commit 38fae3467ccd95c0a4929c3fe6aef69035630de0
Author: Evan <[email protected]>
AuthorDate: Fri Jul 3 19:54:19 2026 +0200
[client] Avoid per-record AutoPartitionStrategy construction in
WriterClient.doSend in log tables (#3529)
* perf(write): skip auto-partition work for non-partitioned tables
Building the AutoPartitionStrategy on every record needlessly cost CPU in
the hot write path, even for tables without partition keys.
Guard the dynamic partition creation call in WriterClient.doSend behind a
partition-keys check so the strategy is only resolved for partitioned tables,
and cache AutoPartitionStrategy lazily in TableConfig via a volatile field
rather than rebuilding it on each access.
Signed-off-by: Evan <[email protected]>
* address comments
- test the strategy is memoized, so repeated calls return the same
instancetestAutoPartitionStrategyIsCached
- use tableInfo.isPartitioned instead of
!tableInfo.getPartitionKeys().isEmpty()
- address comments
Signed-off-by: Evan <[email protected]>
* Address review feedback on #3529
instead of memoizing the strategy on TableConfig, pass TableInfo into
checkAndCreatePartitionAsync and resolve getAutoPartitionStrategy()
inside the create branch. The strategy is only needed when a partition
is actually created, so on the common "already exists" path it is never
built. This removes the cached field
- WriterClient: pass tableInfo instead of pre-computing the strategy
- DynamicPartitionCreator: resolve partitionKeys + strategy in the create
branch only
- TableConfig: drop the cached volatile field and stale javadoc
- Remove the caching test added previously
Signed-off-by: Evan <[email protected]>
---------
Signed-off-by: Evan <[email protected]>
---
.../apache/fluss/client/write/DynamicPartitionCreator.java | 11 +++++++----
.../main/java/org/apache/fluss/client/write/WriterClient.java | 8 ++++----
2 files changed, 11 insertions(+), 8 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java
index 7a392f4af..a6a73354f 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java
@@ -23,6 +23,7 @@ import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.PartitionNotExistException;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.utils.AutoPartitionStrategy;
import org.apache.fluss.utils.ExceptionUtils;
@@ -66,9 +67,7 @@ public class DynamicPartitionCreator {
}
public void checkAndCreatePartitionAsync(
- PhysicalTablePath physicalTablePath,
- List<String> partitionKeys,
- AutoPartitionStrategy autoPartitionStrategy) {
+ PhysicalTablePath physicalTablePath, TableInfo tableInfo) {
String partitionName = physicalTablePath.getPartitionName();
if (partitionName == null) {
// no need to check and create partition
@@ -87,7 +86,11 @@ public class DynamicPartitionCreator {
// if the partition exists, we should skip creating it.
LOG.debug("Partition {} already exists, skipping.",
physicalTablePath);
} else {
- // Validate early, before touching any state.
+ // Validate early, before touching any state. The strategy is
only resolved here,
+ // on the partition-creation path, not on the common "already
exists" path.
+ List<String> partitionKeys = tableInfo.getPartitionKeys();
+ AutoPartitionStrategy autoPartitionStrategy =
+ tableInfo.getTableConfig().getAutoPartitionStrategy();
ResolvedPartitionSpec resolvedPartitionSpec =
ResolvedPartitionSpec.fromPartitionName(partitionKeys,
partitionName);
validateAutoPartitionTime(
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
index b4c96b0ac..ff8b9232f 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
@@ -176,10 +176,10 @@ public class WriterClient {
TableInfo tableInfo = record.getTableInfo();
PhysicalTablePath physicalTablePath =
record.getPhysicalTablePath();
- dynamicPartitionCreator.checkAndCreatePartitionAsync(
- physicalTablePath,
- tableInfo.getPartitionKeys(),
- tableInfo.getTableConfig().getAutoPartitionStrategy());
+ // Skip the call entirely on non-partitioned tables; there is no
partition to create.
+ if (tableInfo.isPartitioned()) {
+
dynamicPartitionCreator.checkAndCreatePartitionAsync(physicalTablePath,
tableInfo);
+ }
// maybe create bucket assigner.
Cluster cluster = metadataUpdater.getCluster();
