Skip to content

[HUDI-6882] Differentiate between replacecommits in cluster planning #9755

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yihua merged 2 commits into from
Sep 21, 2023
Merged

[HUDI-6882] Differentiate between replacecommits in cluster planning #9755

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
71dea53
make sure replacecommit is cluster when getting latest cluster in clu…
Sep 20, 2023
0d495d3
address feedback
Sep 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions ...mmon/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanActionExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,7 @@ public ClusteringPlanActionExecutor(HoodieEngineContext context,

protected Option<HoodieClusteringPlan> createClusteringPlan() {
LOG.info("Checking if clustering needs to be run on " + config.getBasePath());
Option<HoodieInstant> lastClusteringInstant = table.getActiveTimeline()
.filter(s -> s.getAction().equalsIgnoreCase(HoodieTimeline.REPLACE_COMMIT_ACTION)).lastInstant();
Option<HoodieInstant> lastClusteringInstant = table.getActiveTimeline().getLastClusterCommit();

int commitsSinceLastClustering = table.getActiveTimeline().getCommitsTimeline().filterCompletedInstants()
.findInstantsAfter(lastClusteringInstant.map(HoodieInstant::getTimestamp).orElse("0"), Integer.MAX_VALUE)
Expand Down
16 changes: 16 additions & 0 deletions hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,15 @@

package org.apache.hudi.common.table.timeline;

import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.timeline.HoodieInstant.State;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.exception.HoodieException;

import java.io.IOException;
import java.io.Serializable;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
Expand Down Expand Up @@ -490,6 +493,19 @@ public Option<HoodieInstant> getFirstNonSavepointCommit() {
}
return firstNonSavepointCommit;
}

public Option<HoodieInstant> getLastClusterCommit() {
return Option.fromJavaOptional(getCommitsTimeline().filter(s -> s.getAction().equalsIgnoreCase(HoodieTimeline.REPLACE_COMMIT_ACTION))
.getReverseOrderedInstants()
.filter(i -> {
try {
HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(i, this);
Copy link
Contributor

@yihua yihua Mar 22, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This introduced a regression that the pending clustering is no longer considered when figuring out the lastClusteringInstant.

return metadata.getOperationType().equals(WriteOperationType.CLUSTER);
} catch (IOException e) {
return false;
}
}).findFirst());
}

@Override
public Option<byte[]> getInstantDetails(HoodieInstant instant) {
Expand Down
54 changes: 52 additions & 2 deletions ...k-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ package org.apache.hudi.functional

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hudi.DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME
import org.apache.hudi.DataSourceWriteOptions.{INLINE_CLUSTERING_ENABLE, KEYGENERATOR_CLASS_NAME}
import org.apache.hudi.HoodieConversionUtils.toJavaOption
import org.apache.hudi.QuickstartUtils.{convertToStringList, getQuickstartWriteConfigs}
import org.apache.hudi.client.common.HoodieSparkEngineContext
Expand All @@ -28,7 +28,7 @@ import org.apache.hudi.common.config.{HoodieCommonConfig, HoodieMetadataConfig}
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.model.{HoodieRecord, WriteOperationType}
import org.apache.hudi.common.table.timeline.{HoodieInstant, TimelineUtils}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineUtils}
import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
import org.apache.hudi.common.testutils.HoodieTestDataGenerator
import org.apache.hudi.common.testutils.RawTripTestPayload.{deleteRecordsToStrings, recordsToStrings}
Expand Down Expand Up @@ -1724,6 +1724,56 @@ class TestCOWDataSource extends HoodieSparkClientTestBase with ScalaAssertionSup
val metadata = TimelineUtils.getCommitMetadata(latestCommit.get(), timeline)
metadata.getOperationType.equals(WriteOperationType.UPSERT)
}

@ParameterizedTest
@EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO", "SPARK"))
def testInsertOverwriteCluster(recordType: HoodieRecordType): Unit = {
val (writeOpts, _) = getWriterReaderOpts(recordType)

// Insert Operation
val records = recordsToStrings(dataGen.generateInserts("000", 100)).toList
val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))

val optsWithCluster = Map(
INLINE_CLUSTERING_ENABLE.key() -> "true",
"hoodie.clustering.inline.max.commits" -> "2",
"hoodie.clustering.plan.strategy.sort.columns" -> "_row_key",
"hoodie.insert.shuffle.parallelism" -> "4",
"hoodie.upsert.shuffle.parallelism" -> "4",
DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
HoodieWriteConfig.TBL_NAME.key -> "hoodie_test"
) ++ writeOpts
inputDF.write.format("hudi")
.options(optsWithCluster)
.option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
.mode(SaveMode.Overwrite)
.save(basePath)

for (i <- 1 until 6) {
val records = recordsToStrings(dataGen.generateInsertsForPartition("00" + i, 10, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH)).toList
val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
inputDF.write.format("hudi")
.options(optsWithCluster)
.option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OVERWRITE_OPERATION_OPT_VAL)
.mode(SaveMode.Append)
.save(basePath)
}

val metaClient = HoodieTableMetaClient.builder()
.setBasePath(basePath)
.setConf(hadoopConf)
.build()
val timeline = metaClient.getActiveTimeline
val instants = timeline.getAllCommitsTimeline.filterCompletedInstants.getInstants
assertEquals(9, instants.size)
val replaceInstants = instants.filter(i => i.getAction.equals(HoodieTimeline.REPLACE_COMMIT_ACTION)).toList
assertEquals(8, replaceInstants.size)
val clusterInstants = replaceInstants.filter(i => {
TimelineUtils.getCommitMetadata(i, metaClient.getActiveTimeline).getOperationType.equals(WriteOperationType.CLUSTER)
})
assertEquals(3, clusterInstants.size)
}
}

object TestCOWDataSource {
Expand Down