You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by zh...@apache.org on 2025/06/18 06:54:55 UTC
(hudi) branch HUDI-9352 created (now f0025e0e091)
This is an automated email from the ASF dual-hosted git repository.
zhangyue19921010 pushed a change to branch HUDI-9352
in repository https://gitbox.apache.org/repos/asf/hudi.git
at f0025e0e091 Add sort columns in plan generated by Flink
This branch includes the following new commits:
new f0025e0e091 Add sort columns in plan generated by Flink
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(hudi) 01/01: Add sort columns in plan generated by Flink
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangyue19921010 pushed a commit to branch HUDI-9352
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit f0025e0e091cde621ed484b656843e1537f12e7c
Author: YueZhang <zh...@163.com>
AuthorDate: Wed Jun 18 14:54:37 2025 +0800
Add sort columns in plan generated by Flink
---
.../org/apache/hudi/util/FlinkWriteClients.java | 1 +
.../apache/hudi/sink/ITTestDataStreamWrite.java | 30 +++++++++++++++++++++-
2 files changed, 30 insertions(+), 1 deletion(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java
index 6e0b26a4f23..70929c142d2 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java
@@ -180,6 +180,7 @@ public class FlinkWriteClients {
.withClusteringPartitionRegexPattern(conf.get(FlinkOptions.CLUSTERING_PLAN_STRATEGY_PARTITION_REGEX_PATTERN))
.withClusteringPartitionSelected(conf.get(FlinkOptions.CLUSTERING_PLAN_STRATEGY_PARTITION_SELECTED))
.withAsyncClusteringMaxCommits(conf.getInteger(FlinkOptions.CLUSTERING_DELTA_COMMITS))
+ .withClusteringSortColumns(conf.getString(FlinkOptions.CLUSTERING_SORT_COLUMNS))
.withScheduleInlineClustering(conf.getBoolean(FlinkOptions.CLUSTERING_SCHEDULE_ENABLED))
.withAsyncClustering(conf.getBoolean(FlinkOptions.CLUSTERING_ASYNC_ENABLED))
.build())
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java
index 8aca31b2173..00336776c77 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java
@@ -18,9 +18,15 @@
package org.apache.hudi.sink;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
import org.apache.hudi.client.model.HoodieFlinkInternalRow;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.configuration.OptionsInference;
@@ -29,6 +35,7 @@ import org.apache.hudi.sink.transform.ChainedTransformer;
import org.apache.hudi.sink.transform.Transformer;
import org.apache.hudi.sink.utils.Pipelines;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.table.catalog.HoodieCatalog;
import org.apache.hudi.table.catalog.TableOptionProperties;
import org.apache.hudi.util.AvroSchemaConverter;
@@ -74,10 +81,13 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
import static org.apache.hudi.config.HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE;
import static org.apache.hudi.config.HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP;
import static org.apache.hudi.table.catalog.CatalogOptions.CATALOG_PATH;
import static org.apache.hudi.table.catalog.CatalogOptions.DEFAULT_DATABASE;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Integration test for Flink Hoodie stream sink.
@@ -171,7 +181,8 @@ public class ITTestDataStreamWrite extends TestLogger {
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testWriteCopyOnWriteWithClustering(boolean sortClusteringEnabled) throws Exception {
- Configuration conf = TestConfigurations.getDefaultConf(tempFile.toURI().toString());
+ String basePath = tempFile.toURI().toString();
+ Configuration conf = TestConfigurations.getDefaultConf(basePath);
conf.setBoolean(FlinkOptions.CLUSTERING_SCHEDULE_ENABLED, true);
conf.setInteger(FlinkOptions.CLUSTERING_DELTA_COMMITS, 1);
conf.setString(FlinkOptions.OPERATION, "insert");
@@ -180,6 +191,23 @@ public class ITTestDataStreamWrite extends TestLogger {
}
writeWithClusterAndCheckExpected(conf, "cow_write_with_cluster", 1, EXPECTED);
+ if (sortClusteringEnabled) {
+ HadoopStorageConfiguration storageConf = new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf));
+ HoodieTableMetaClient metaClient =
+ HoodieTestUtils.createMetaClient(storageConf, basePath);
+ HoodieInstant clusteringInstant = metaClient.getActiveTimeline().getLastClusteringInstant().get();
+ Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
+ metaClient, clusteringInstant);
+ assertTrue(clusteringPlanOption.isPresent());
+ HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
+ Map<String, String> strategyParams = clusteringPlan.getStrategy().getStrategyParams();
+ // could be used in spark MultipleSparkJobExecutionStrategy
+ Option<String[]> orderByColumnsOpt =
+ Option.ofNullable(strategyParams.get(PLAN_STRATEGY_SORT_COLUMNS.key()))
+ .map(listStr -> listStr.split(","));
+ assertTrue(orderByColumnsOpt.isPresent());
+ assertTrue(orderByColumnsOpt.get()[0].equalsIgnoreCase("uuid"));
+ }
}
@ParameterizedTest