You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2022/06/02 01:48:55 UTC
[hudi] branch master updated: [HUDI-4167] Remove the timeline refresh with initializing hoodie table (#5716)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 7f8630cc57 [HUDI-4167] Remove the timeline refresh with initializing hoodie table (#5716)
7f8630cc57 is described below
commit 7f8630cc57fbb9d29e8dc7ca87b582264da073fd
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Thu Jun 2 09:48:48 2022 +0800
[HUDI-4167] Remove the timeline refresh with initializing hoodie table (#5716)
The timeline refresh on table initialization invokes the fs view #sync, which has two actions now:
1. reload the timeline of the fs view, so that the next fs view request is based on this timeline metadata
2. if this is a local fs view, clear all the local states; if this is a remote fs view, send request to sync the remote fs view
But, let's see the construction, the meta client is instantiated freshly so the timeline is already the latest,
the table is also constructed freshly, so the fs view has no local states, that means, the #sync is unnecessary totally.
In this patch, the metadata lifecycle and data set fs view are kept in sync, when the fs view is refreshed, the underneath metadata
is also refreshed synchronouly. The freshness of the metadata follows the same rules as data fs view:
1. if the fs view is local, the visibility is based on the client table metadata client's latest commit
2. if the fs view is remote, the timeline server would #sync the fs view and metadata together based on the lagging server local timeline
From the perspective of client, no need to care about the refresh action anymore no matter whether the metadata table is enabled or not.
That make the client logic more clear and less error-prone.
Removes the timeline refresh has another benefit: if avoids unncecessary #refresh of the remote fs view, if all the clients send request to #sync the
remote fs view, the server would encounter conflicts and the client encounters a response error.
---
.../org/apache/hudi/cli/commands/SparkMain.java | 2 +-
.../apache/hudi/client/BaseHoodieWriteClient.java | 28 ++++++++++------------
.../org/apache/hudi/config/HoodieWriteConfig.java | 9 +++++--
.../apache/hudi/client/HoodieFlinkWriteClient.java | 3 +--
.../org/apache/hudi/table/HoodieFlinkTable.java | 10 --------
.../index/bloom/TestFlinkHoodieBloomIndex.java | 2 +-
.../apache/hudi/client/HoodieJavaWriteClient.java | 4 +---
.../apache/hudi/client/SparkRDDWriteClient.java | 12 ++++------
.../org/apache/hudi/table/HoodieSparkTable.java | 17 +------------
.../hudi/client/TestTableSchemaEvolution.java | 8 ++++++-
.../functional/TestHoodieBackedTableMetadata.java | 2 +-
.../hudi/table/TestHoodieMergeOnReadTable.java | 2 +-
.../hudi/testutils/HoodieClientTestHarness.java | 2 +-
.../table/view/RocksDbBasedFileSystemView.java | 1 +
.../hudi/metadata/HoodieBackedTableMetadata.java | 21 ++++++++++++----
.../hudi/timeline/service/TimelineService.java | 3 ++-
16 files changed, 59 insertions(+), 67 deletions(-)
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
index 9fe83f1995..43fe168587 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
@@ -296,7 +296,7 @@ public class SparkMain {
SparkRDDWriteClient client = createHoodieClient(jsc, basePath, false);
HoodieWriteConfig config = client.getConfig();
HoodieEngineContext context = client.getEngineContext();
- HoodieSparkTable table = HoodieSparkTable.create(config, context, true);
+ HoodieSparkTable table = HoodieSparkTable.create(config, context);
WriteMarkersFactory.get(config.getMarkersType(), table, instantTime)
.quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
return 0;
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index 5c485bed05..455cb644c7 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -296,11 +296,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
}
}
- protected HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf) {
- return createTable(config, hadoopConf, false);
- }
-
- protected abstract HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf, boolean refreshTimeline);
+ protected abstract HoodieTable<T, I, K, O> createTable(HoodieWriteConfig config, Configuration hadoopConf);
void emitCommitMetrics(String instantTime, HoodieCommitMetadata metadata, String actionType) {
try {
@@ -365,7 +361,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
*/
protected void rollbackFailedBootstrap() {
LOG.info("Rolling back pending bootstrap if present");
- HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
+ HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
HoodieTimeline inflightTimeline = table.getMetaClient().getCommitsTimeline().filterPendingExcludingCompaction();
Option<String> instant = Option.fromJavaOptional(
inflightTimeline.getReverseOrderedInstants().map(HoodieInstant::getTimestamp).findFirst());
@@ -634,7 +630,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
* Run any pending compactions.
*/
public void runAnyPendingCompactions() {
- runAnyPendingCompactions(createTable(config, hadoopConf, config.isMetadataTableEnabled()));
+ runAnyPendingCompactions(createTable(config, hadoopConf));
}
/**
@@ -644,7 +640,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
* @param comment - Comment for the savepoint
*/
public void savepoint(String user, String comment) {
- HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
+ HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
if (table.getCompletedCommitsTimeline().empty()) {
throw new HoodieSavepointException("Could not savepoint. Commit timeline is empty");
}
@@ -668,7 +664,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
* @param comment - Comment for the savepoint
*/
public void savepoint(String instantTime, String user, String comment) {
- HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
+ HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
table.savepoint(context, instantTime, user, comment);
}
@@ -680,7 +676,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
* @return true if the savepoint was deleted successfully
*/
public void deleteSavepoint(String savepointTime) {
- HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
+ HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
SavepointHelpers.deleteSavepoint(table, savepointTime);
}
@@ -1012,7 +1008,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
*/
public Option<String> scheduleIndexing(List<MetadataPartitionType> partitionTypes) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
- Option<HoodieIndexPlan> indexPlan = createTable(config, hadoopConf, config.isMetadataTableEnabled())
+ Option<HoodieIndexPlan> indexPlan = createTable(config, hadoopConf)
.scheduleIndexing(context, instantTime, partitionTypes);
return indexPlan.isPresent() ? Option.of(instantTime) : Option.empty();
}
@@ -1024,7 +1020,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
* @return {@link Option<HoodieIndexCommitMetadata>} after successful indexing.
*/
public Option<HoodieIndexCommitMetadata> index(String indexInstantTime) {
- return createTable(config, hadoopConf, config.isMetadataTableEnabled()).index(context, indexInstantTime);
+ return createTable(config, hadoopConf).index(context, indexInstantTime);
}
/**
@@ -1339,17 +1335,17 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
return Option.empty();
case CLUSTER:
LOG.info("Scheduling clustering at instant time :" + instantTime);
- Option<HoodieClusteringPlan> clusteringPlan = createTable(config, hadoopConf, config.isMetadataTableEnabled())
+ Option<HoodieClusteringPlan> clusteringPlan = createTable(config, hadoopConf)
.scheduleClustering(context, instantTime, extraMetadata);
return clusteringPlan.isPresent() ? Option.of(instantTime) : Option.empty();
case COMPACT:
LOG.info("Scheduling compaction at instant time :" + instantTime);
- Option<HoodieCompactionPlan> compactionPlan = createTable(config, hadoopConf, config.isMetadataTableEnabled())
+ Option<HoodieCompactionPlan> compactionPlan = createTable(config, hadoopConf)
.scheduleCompaction(context, instantTime, extraMetadata);
return compactionPlan.isPresent() ? Option.of(instantTime) : Option.empty();
case CLEAN:
LOG.info("Scheduling cleaning at instant time :" + instantTime);
- Option<HoodieCleanerPlan> cleanerPlan = createTable(config, hadoopConf, config.isMetadataTableEnabled())
+ Option<HoodieCleanerPlan> cleanerPlan = createTable(config, hadoopConf)
.scheduleCleaning(context, instantTime, extraMetadata);
return cleanerPlan.isPresent() ? Option.of(instantTime) : Option.empty();
default:
@@ -1702,6 +1698,6 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
// try to save history schemas
FileBasedInternalSchemaStorageManager schemasManager = new FileBasedInternalSchemaStorageManager(metaClient);
schemasManager.persistHistorySchemaStr(instantTime, SerDeHelper.inheritSchemas(newSchema, historySchemaStr));
- commitStats(instantTime, Collections.EMPTY_LIST, Option.of(extraMeta), commitActionType);
+ commitStats(instantTime, Collections.emptyList(), Option.of(extraMeta), commitActionType);
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index 42208a0734..1603965ea9 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -364,8 +364,8 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty<Boolean> REFRESH_TIMELINE_SERVER_BASED_ON_LATEST_COMMIT = ConfigProperty
.key("hoodie.refresh.timeline.server.based.on.latest.commit")
- .defaultValue(false)
- .withDocumentation("Refresh timeline in timeline server based on latest commit apart from timeline hash difference. By default (false), ");
+ .defaultValue(true)
+ .withDocumentation("Refresh timeline in timeline server based on latest commit apart from timeline hash difference. By default (true).");
public static final ConfigProperty<Long> INITIAL_CONSISTENCY_CHECK_INTERVAL_MS = ConfigProperty
.key("hoodie.consistency.check.initial_interval_ms")
@@ -2499,6 +2499,11 @@ public class HoodieWriteConfig extends HoodieConfig {
return this;
}
+ public Builder withRefreshTimelineServerBasedOnLatestCommit(boolean refreshTimelineServerBasedOnLatestCommit) {
+ writeConfig.setValue(REFRESH_TIMELINE_SERVER_BASED_ON_LATEST_COMMIT, Boolean.toString(refreshTimelineServerBasedOnLatestCommit));
+ return this;
+ }
+
protected void setDefaults() {
writeConfig.setDefaultValue(MARKERS_TYPE, getDefaultMarkersType(engineType));
// Check for mandatory properties
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
index ddfbabaf36..b68cf97e9a 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -117,8 +117,7 @@ public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
}
@Override
- protected HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf,
- boolean refreshTimeline) {
+ protected HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf) {
return HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) context);
}
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkTable.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkTable.java
index 6eae15e7e1..26149918c6 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkTable.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkTable.java
@@ -62,13 +62,6 @@ public abstract class HoodieFlinkTable<T extends HoodieRecordPayload>
public static <T extends HoodieRecordPayload> HoodieFlinkTable<T> create(HoodieWriteConfig config,
HoodieFlinkEngineContext context,
HoodieTableMetaClient metaClient) {
- return HoodieFlinkTable.create(config, context, metaClient, config.isMetadataTableEnabled());
- }
-
- public static <T extends HoodieRecordPayload> HoodieFlinkTable<T> create(HoodieWriteConfig config,
- HoodieFlinkEngineContext context,
- HoodieTableMetaClient metaClient,
- boolean refreshTimeline) {
final HoodieFlinkTable<T> hoodieFlinkTable;
switch (metaClient.getTableType()) {
case COPY_ON_WRITE:
@@ -80,9 +73,6 @@ public abstract class HoodieFlinkTable<T extends HoodieRecordPayload>
default:
throw new HoodieException("Unsupported table type :" + metaClient.getTableType());
}
- if (refreshTimeline) {
- hoodieFlinkTable.getHoodieView().sync();
- }
return hoodieFlinkTable;
}
diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java
index 50adabbd58..e23ee4ad58 100644
--- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java
+++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java
@@ -104,7 +104,7 @@ public class TestFlinkHoodieBloomIndex extends HoodieFlinkClientTestHarness {
public void testLoadInvolvedFiles(boolean rangePruning, boolean treeFiltering, boolean bucketizedChecking) throws Exception {
HoodieWriteConfig config = makeConfig(rangePruning, treeFiltering, bucketizedChecking);
HoodieBloomIndex index = new HoodieBloomIndex(config, ListBasedHoodieBloomIndexHelper.getInstance());
- HoodieTable hoodieTable = HoodieFlinkTable.create(config, context, metaClient, false);
+ HoodieTable hoodieTable = HoodieFlinkTable.create(config, context, metaClient);
HoodieFlinkWriteableTestTable testTable = HoodieFlinkWriteableTestTable.of(hoodieTable, SCHEMA);
// Create some partitions, and put some files
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
index 7f5dc19baf..fbfb85bab3 100644
--- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
+++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
@@ -89,9 +89,7 @@ public class HoodieJavaWriteClient<T extends HoodieRecordPayload> extends
}
@Override
- protected HoodieTable createTable(HoodieWriteConfig config,
- Configuration hadoopConf,
- boolean refreshTimeline) {
+ protected HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf) {
return HoodieJavaTable.create(config, context);
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
index 7f9ec05e3c..fe6ea975e3 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -123,10 +123,8 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
}
@Override
- protected HoodieTable createTable(HoodieWriteConfig config,
- Configuration hadoopConf,
- boolean refreshTimeline) {
- return HoodieSparkTable.create(config, context, refreshTimeline);
+ protected HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf) {
+ return HoodieSparkTable.create(config, context);
}
@Override
@@ -333,7 +331,7 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
@Override
protected HoodieWriteMetadata<JavaRDD<WriteStatus>> compact(String compactionInstantTime, boolean shouldComplete) {
- HoodieSparkTable<T> table = HoodieSparkTable.create(config, context, true);
+ HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
preWrite(compactionInstantTime, WriteOperationType.COMPACT, table.getMetaClient());
HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
@@ -352,7 +350,7 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
@Override
public HoodieWriteMetadata<JavaRDD<WriteStatus>> cluster(String clusteringInstant, boolean shouldComplete) {
- HoodieSparkTable<T> table = HoodieSparkTable.create(config, context, config.isMetadataTableEnabled());
+ HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
preWrite(clusteringInstant, WriteOperationType.CLUSTER, table.getMetaClient());
HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingReplaceTimeline();
HoodieInstant inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(clusteringInstant);
@@ -434,7 +432,7 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
}
// Create a Hoodie table which encapsulated the commits and files visible
- return HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient, config.isMetadataTableEnabled());
+ return HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient);
}
/**
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java
index 20e3bd4c14..66d51c9128 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java
@@ -54,30 +54,18 @@ public abstract class HoodieSparkTable<T extends HoodieRecordPayload>
}
public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config, HoodieEngineContext context) {
- return create(config, context, false);
- }
-
- public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config, HoodieEngineContext context,
- boolean refreshTimeline) {
HoodieTableMetaClient metaClient =
HoodieTableMetaClient.builder().setConf(context.getHadoopConf().get()).setBasePath(config.getBasePath())
.setLoadActiveTimelineOnLoad(true).setConsistencyGuardConfig(config.getConsistencyGuardConfig())
.setLayoutVersion(Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion())))
.setFileSystemRetryConfig(config.getFileSystemRetryConfig())
.setProperties(config.getProps()).build();
- return HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient, refreshTimeline);
+ return HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient);
}
public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config,
HoodieSparkEngineContext context,
HoodieTableMetaClient metaClient) {
- return create(config, context, metaClient, false);
- }
-
- public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config,
- HoodieSparkEngineContext context,
- HoodieTableMetaClient metaClient,
- boolean refreshTimeline) {
HoodieSparkTable<T> hoodieSparkTable;
switch (metaClient.getTableType()) {
case COPY_ON_WRITE:
@@ -89,9 +77,6 @@ public abstract class HoodieSparkTable<T extends HoodieRecordPayload>
default:
throw new HoodieException("Unsupported table type :" + metaClient.getTableType());
}
- if (refreshTimeline) {
- hoodieSparkTable.getHoodieView().sync();
- }
return hoodieSparkTable;
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestTableSchemaEvolution.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestTableSchemaEvolution.java
index 1cb7bcbfc4..98bcb11033 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestTableSchemaEvolution.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestTableSchemaEvolution.java
@@ -515,7 +515,13 @@ public class TestTableSchemaEvolution extends HoodieClientTestBase {
return getConfigBuilder(schema)
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(IndexType.INMEMORY).build())
.withCompactionConfig(HoodieCompactionConfig.newBuilder().withMaxNumDeltaCommitsBeforeCompaction(1).build())
- .withAvroSchemaValidate(true);
+ .withAvroSchemaValidate(true)
+ // The test has rollback instants on the timeline,
+ // these rollback instants use real time as instant time, whose instant time is always greater than
+ // the normal commits instant time, this breaks the refresh rule introduced in HUDI-2761:
+ // The last client instant is always the rollback instant but not the normal commit.
+ // Always refresh the timeline when client and server have different timeline.
+ .withRefreshTimelineServerBasedOnLatestCommit(false);
}
@Override
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java
index 9a8fc55a20..e19c8fc1a2 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java
@@ -111,7 +111,7 @@ public class TestHoodieBackedTableMetadata extends TestHoodieMetadataBase {
assertEquals(fsPartitions, metadataPartitions, "Partitions should match");
// Files within each partition should match
- HoodieTable table = HoodieSparkTable.create(writeConfig, context, true);
+ HoodieTable table = HoodieSparkTable.create(writeConfig, context);
TableFileSystemView tableView = table.getHoodieView();
List<String> fullPartitionPaths = fsPartitions.stream().map(partition -> basePath + "/" + partition).collect(Collectors.toList());
Map<String, FileStatus[]> partitionToFilesMap = tableMetadata.getAllFilesInPartitions(fullPartitionPaths);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
index b9f025223b..0ce6ca0ee9 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
@@ -247,7 +247,7 @@ public class TestHoodieMergeOnReadTable extends SparkClientFunctionalTestHarness
assertEquals(allPartitions.size(), testTable.listAllBaseFiles().length);
// Verify that all data file has one log file
- HoodieTable table = HoodieSparkTable.create(config, context(), metaClient, true);
+ HoodieTable table = HoodieSparkTable.create(config, context(), metaClient);
for (String partitionPath : dataGen.getPartitionPaths()) {
List<FileSlice> groupedLogFiles =
table.getSliceView().getLatestFileSlices(partitionPath).collect(Collectors.toList());
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestHarness.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestHarness.java
index 4504c552c9..d0365dced1 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestHarness.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestHarness.java
@@ -559,7 +559,7 @@ public abstract class HoodieClientTestHarness extends HoodieCommonTestHarness im
// Files within each partition should match
metaClient = HoodieTableMetaClient.reload(metaClient);
- HoodieTable table = HoodieSparkTable.create(writeConfig, engineContext, true);
+ HoodieTable table = HoodieSparkTable.create(writeConfig, engineContext);
TableFileSystemView tableView = table.getHoodieView();
List<String> fullPartitionPaths = fsPartitions.stream().map(partition -> basePath + "/" + partition).collect(Collectors.toList());
Map<String, FileStatus[]> partitionToFilesMap = tableMetadata.getAllFilesInPartitions(fullPartitionPaths);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java
index af0dc13016..02a406e7e0 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java
@@ -199,6 +199,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste
LOG.info("Deleting all rocksdb data associated with table filesystem view");
rocksDB.close();
rocksDB = new RocksDBDAO(metaClient.getBasePath(), config.getRocksdbBasePath());
+ schemaHelper.getAllColumnFamilies().forEach(rocksDB::addColumnFamily);
}
@Override
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
index cf941bb70c..e8937b39dc 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
@@ -550,10 +550,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
@Override
public void close() {
- for (Pair<String, String> partitionFileSlicePair : partitionReaders.keySet()) {
- close(partitionFileSlicePair);
- }
- partitionReaders.clear();
+ closePartitionReaders();
}
/**
@@ -567,6 +564,16 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
closeReader(readers);
}
+ /**
+ * Close and clear all the partitions readers.
+ */
+ private void closePartitionReaders() {
+ for (Pair<String, String> partitionFileSlicePair : partitionReaders.keySet()) {
+ close(partitionFileSlicePair);
+ }
+ partitionReaders.clear();
+ }
+
private void closeReader(Pair<HoodieFileReader, HoodieMetadataMergedLogRecordReader> readers) {
if (readers != null) {
try {
@@ -624,5 +631,11 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
public void reset() {
initIfNeeded();
dataMetaClient.reloadActiveTimeline();
+ if (metadataMetaClient != null) {
+ metadataMetaClient.reloadActiveTimeline();
+ }
+ // the cached reader has max instant time restriction, they should be cleared
+ // because the metadata timeline may have changed.
+ closePartitionReaders();
}
}
diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/TimelineService.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/TimelineService.java
index 40669f50e4..2ff2168221 100644
--- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/TimelineService.java
+++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/TimelineService.java
@@ -150,7 +150,7 @@ public class TimelineService {
private int markerBatchNumThreads = 20;
private long markerBatchIntervalMs = 50L;
private int markerParallelism = 100;
- private boolean refreshTimelineBasedOnLatestCommit = false;
+ private boolean refreshTimelineBasedOnLatestCommit = true;
public Builder() {
}
@@ -240,6 +240,7 @@ public class TimelineService {
config.markerBatchNumThreads = this.markerBatchNumThreads;
config.markerBatchIntervalMs = this.markerBatchIntervalMs;
config.markerParallelism = this.markerParallelism;
+ config.refreshTimelineBasedOnLatestCommit = this.refreshTimelineBasedOnLatestCommit;
return config;
}
}