binary-signal opened a new issue, #3527: URL: https://github.com/apache/fluss/issues/3527
### Search before asking - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar. ### Fluss version main (development) ### Please describe the bug 🐞 ### Fluss version main (development) ### Please describe the bug A Flink job sinking to a non-partitioned Fluss append-only log table backpressures the upstream pipeline. Profiling the sink subtask shows that roughly half of its CPU is spent on auto-partition strategy construction for a table that isn't partitioned in the first place. #### Table ```sql CREATE TABLE `fluss_catalog`.`cdstream_db`.`cluster_combinations` (`combinationId` VARCHAR(2147483647), `leftClusters` ARRAY<INT>, `rightClusters` ARRAY<INT>, `level` INT, `size` BIGINT, `decisive` BOOLEAN, `positive` BOOLEAN, `bounded` BOOLEAN, `singleton` BOOLEAN, `parentId` VARCHAR(2147483647), `boundingTimestamp` TIMESTAMP(3), `discoveredTimestamp` TIMESTAMP(3), `lowerBound` DOUBLE, `upperBound` DOUBLE) WITH ( 'bucket.num' = '64', 'table.log.format' = 'arrow', 'table.log.ttl' = '1h', 'table.replication.factor' = '1', 'table.datalake.enabled' = 'false' ); ``` #### Writer sink flamegraph <img width="1960" height="1450" alt="Image" src="https://github.com/user-attachments/assets/3f752699-9515-4e82-a0e8-7d7e6b88ce55"; /> | Code path | Samples | % | |---------------------------------------------------------------------------------|--------:|-----:| | `java.util.TimeZone.getTimeZone(String)` | 331 | 41.4 | | `AutoPartitionStrategy.from` + `TableConfig.getAutoPartitionStrategy` around it | 71 | 8.9 | | Arrow row encoding | 136 | 17.0 | | User serializer | 93 | 11.6 | | ZSTD compression in batch close | 49 | 6.1 | | Everything else (sink, bucket assignment, RPC, IO) | <50 | <6 | About 50% of the writer thread's CPU is in `AutoPartitionStrategy.from(...)`, dominated by the JDK's `TimeZone.getTimeZone(String)`. That JDK method walks a global static registry under a synchronized block and clones the returned object every call, so it is genuinely slow to invoke at high frequency. #### Call chain ``` WriterClient.doSend (line 182) dynamicPartitionCreator.checkAndCreatePartitionAsync( physicalTablePath, tableInfo.getPartitionKeys(), tableInfo.getTableConfig().getAutoPartitionStrategy()) // per record TableConfig.getAutoPartitionStrategy (line 161) return AutoPartitionStrategy.from(config); // new object each call AutoPartitionStrategy.from(Configuration) (line 57) TimeZone.getTimeZone( (line 64) conf.getString(TABLE_AUTO_PARTITION_TIMEZONE)) // 41% of CPU ``` #### Source `fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java:179-182` — called once per record: ```java dynamicPartitionCreator.checkAndCreatePartitionAsync( physicalTablePath, tableInfo.getPartitionKeys(), tableInfo. getTableConfig(). getAutoPartitionStrategy()); ``` `fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java:159-162` — uncached, builds a fresh strategy every call: ```java /** Gets the auto partition strategy of the table. */ public AutoPartitionStrategy getAutoPartitionStrategy() { return AutoPartitionStrategy.from(config); } ``` `fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java:57-65`: ```java public static AutoPartitionStrategy from(Configuration conf) { return new AutoPartitionStrategy( conf.getBoolean(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED), conf.getString(ConfigOptions.TABLE_AUTO_PARTITION_KEY), conf.get(ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT), conf.getInt(ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE), conf.getInt(ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION), TimeZone.getTimeZone(conf.getString(ConfigOptions.TABLE_AUTO_PARTITION_TIMEZONE))); } ``` The callee, `DynamicPartitionCreator.checkAndCreatePartitionAsync` ( `fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java:68-113`), discards the `autoPartitionStrategy` argument for non-partitioned tables and only ever reads it on first write to a partition for partitioned tables: ```java public void checkAndCreatePartitionAsync( PhysicalTablePath physicalTablePath, List<String> partitionKeys, AutoPartitionStrategy autoPartitionStrategy) { String partitionName = physicalTablePath.getPartitionName(); if (partitionName == null) { // no need to check and create partition return; // argument never read } Optional<Long> partitionIdOpt = metadataUpdater.getPartitionId(physicalTablePath); if (!partitionIdOpt.isPresent()) { if (inflightPartitionsToCreate.contains(physicalTablePath)) { ... } else if (forceCheckPartitionExist(physicalTablePath)) { ... } else { ResolvedPartitionSpec resolvedPartitionSpec = ResolvedPartitionSpec.fromPartitionName(partitionKeys, partitionName); validateAutoPartitionTime( resolvedPartitionSpec.toPartitionSpec(), partitionKeys, autoPartitionStrategy); // the only reader ... } } } ``` So in steady state: - Non-partitioned table: strategy used 0 times out of N records. - Partitioned table: strategy used at most P times out of N records, where P is the number of distinct partitions ever written. For long-running streaming jobs P/N is negligible. The strategy is still constructed N times out of N because Java evaluates method arguments before the call. Pure waste for non-partitioned tables, and roughly N/P times too often for partitioned ones. Fixing it does not change behaviour; it only saves CPU. #### When this regressed The `TimeZone.getTimeZone(String)` call has lived inside `AutoPartitionStrategy.from(...)` since the initial commit (`a1280c68`, 2024-11-25). At that point nothing called it on the hot path. The regression came in with #1002, "Support dynamically create partition when writing" (`abb3b86a`, 2025-06-05). That PR added the `dynamicPartitionCreator.checkAndCreatePartitionAsync(..., tableInfo.getTableConfig().getAutoPartitionStrategy())` call inside `WriterClient.doSend` and so promoted `getAutoPartitionStrategy()` from a metadata-path call into an O(records) hot-path call. #### How to reproduce Any Flink job that writes at a non-trivial record rate (we saw it at ~2.5M records/s, 16-way parallel writer) into a Fluss table without a `PARTITIONED BY` clause should reproduce. The flamegraph above came from async-profiler attached to the Flink TaskManager, filtered to the writer subtask thread. ### Are you willing to submit a PR? - [X] I'm willing to submit a PR! ### Solution _No response_ ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
