fix

Author: LsomeYeah

docs/layouts/shortcodes/generated/core_configuration.html

@@ -201,7 +201,7 @@
             <td>Whether enable incremental clustering.</td>
         </tr>
         <tr>
-            <td><h5>clustering.pre-write.enabled</h5></td>
+            <td><h5>clustering.incremental.optimize-write</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
             <td>Whether enable perform clustering before write phase when incremental clustering is enabled.</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2071,8 +2071,8 @@ public InlineElement getDescription() {
                             "The duration after which a partition without new updates is considered a historical partition. "
                                     + "Historical partitions will be automatically fully clustered during the cluster operation.");
 
-    public static final ConfigOption<Boolean> CLUSTERING_PRE_WRITE_ENABLED =
-            key("clustering.pre-write.enabled")
+    public static final ConfigOption<Boolean> CLUSTERING_INCREMENTAL_OPTIMIZE_WRITE =
+            key("clustering.incremental.optimize-write")
                     .booleanType()
                     .defaultValue(false)
                     .withDescription(
@@ -3410,7 +3410,7 @@ public boolean clusteringIncrementalEnabled() {
     }
 
     public boolean preClusteringEnabled() {
-        return options.get(CLUSTERING_PRE_WRITE_ENABLED);
+        return options.get(CLUSTERING_INCREMENTAL_OPTIMIZE_WRITE);
     }
 
     public boolean bucketClusterEnabled() {

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java

@@ -41,7 +41,7 @@
 import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
 import static org.apache.paimon.CoreOptions.CLUSTERING_COLUMNS;
 import static org.apache.paimon.CoreOptions.CLUSTERING_INCREMENTAL;
-import static org.apache.paimon.CoreOptions.CLUSTERING_PRE_WRITE_ENABLED;
+import static org.apache.paimon.CoreOptions.CLUSTERING_INCREMENTAL_OPTIMIZE_WRITE;
 import static org.apache.paimon.CoreOptions.CLUSTERING_STRATEGY;
 import static org.apache.paimon.CoreOptions.MERGE_ENGINE;
 import static org.apache.paimon.flink.FlinkConnectorOptions.CLUSTERING_SAMPLE_FACTOR;
@@ -124,7 +124,7 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
                                     dataStream.getExecutionEnvironment(),
                                     dataStream.getTransformation()));
                     if (!conf.get(CLUSTERING_INCREMENTAL)
-                            || conf.get(CLUSTERING_PRE_WRITE_ENABLED)) {
+                            || conf.get(CLUSTERING_INCREMENTAL_OPTIMIZE_WRITE)) {
                         builder.clusteringIfPossible(
                                 conf.get(CLUSTERING_COLUMNS),
                                 conf.get(CLUSTERING_STRATEGY),

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RangePartitionAndSortForAppendTableITCase.java

@@ -258,7 +258,7 @@ public void testClusteringPreWriteEnabled() throws Exception {
         batchSql(
                 "INSERT INTO test_table /*+ OPTIONS('sink.clustering.by-columns' = 'col1', "
                         + "'sink.parallelism' = '10', 'sink.clustering.strategy' = 'zorder', "
-                        + "'clustering.incremental' = 'true', 'clustering.pre-write.enabled' = 'true') */ "
+                        + "'clustering.incremental' = 'true', 'clustering.incremental.optimize-write' = 'true') */ "
                         + "SELECT * FROM test_source");
         List<Row> sinkRows = batchSql("SELECT * FROM test_table");
         assertThat(sinkRows.size()).isEqualTo(SINK_ROW_NUMBER);