> getPendingRollbackInfos
return infoMap;
}
+ /**
+ * Rolls back the failed delta commits corresponding to the indexing action.
+ * Such delta commits are identified based on the suffix `METADATA_INDEXER_TIME_SUFFIX` ("004").
+ *
+ * TODO(HUDI-5733): This should be cleaned up once the proper fix of rollbacks
+ * in the metadata table is landed.
+ *
+ * @return {@code true} if rollback happens; {@code false} otherwise.
+ */
+ protected boolean rollbackFailedIndexingCommits() {
+ HoodieTable table = createTable(config, hadoopConf);
+ List instantsToRollback = getFailedIndexingCommitsToRollback(table.getMetaClient());
+ Map> pendingRollbacks = getPendingRollbackInfos(table.getMetaClient());
+ instantsToRollback.forEach(entry -> pendingRollbacks.putIfAbsent(entry, Option.empty()));
+ rollbackFailedWrites(pendingRollbacks);
+ return !pendingRollbacks.isEmpty();
+ }
+
+ protected List getFailedIndexingCommitsToRollback(HoodieTableMetaClient metaClient) {
+ Stream inflightInstantsStream = metaClient.getCommitsTimeline()
+ .filter(instant -> !instant.isCompleted()
+ && isIndexingCommit(instant.getTimestamp()))
+ .getInstantsAsStream();
+ return inflightInstantsStream.filter(instant -> {
+ try {
+ return heartbeatClient.isHeartbeatExpired(instant.getTimestamp());
+ } catch (IOException io) {
+ throw new HoodieException("Failed to check heartbeat for instant " + instant, io);
+ }
+ }).map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+ }
+
/**
* Rollback all failed writes.
+ *
* @return true if rollback was triggered. false otherwise.
*/
protected Boolean rollbackFailedWrites() {
@@ -699,6 +737,19 @@ protected List getInstantsToRollback(HoodieTableMetaClient metaClient, H
Stream inflightInstantsStream = getInflightTimelineExcludeCompactionAndClustering(metaClient)
.getReverseOrderedInstants();
if (cleaningPolicy.isEager()) {
+ // Metadata table uses eager cleaning policy, but we need to exclude inflight delta commits
+ // from the async indexer (`HoodieIndexer`).
+ // TODO(HUDI-5733): This should be cleaned up once the proper fix of rollbacks in the
+ // metadata table is landed.
+ if (HoodieTableMetadata.isMetadataTable(metaClient.getBasePathV2().toString())) {
+ return inflightInstantsStream.map(HoodieInstant::getTimestamp).filter(entry -> {
+ if (curInstantTime.isPresent()) {
+ return !entry.equals(curInstantTime.get());
+ } else {
+ return !isIndexingCommit(entry);
+ }
+ }).collect(Collectors.toList());
+ }
return inflightInstantsStream.map(HoodieInstant::getTimestamp).filter(entry -> {
if (curInstantTime.isPresent()) {
return !entry.equals(curInstantTime.get());
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 c3260914bd51b..3fe0fdd330f0b 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
@@ -491,7 +491,7 @@ public abstract O bulkInsertPreppedRecords(I preppedRecords, final String instan
public void preWrite(String instantTime, WriteOperationType writeOperationType,
HoodieTableMetaClient metaClient) {
setOperationType(writeOperationType);
- this.lastCompletedTxnAndMetadata = txnManager.isOptimisticConcurrencyControlEnabled()
+ this.lastCompletedTxnAndMetadata = txnManager.isLockRequired()
? TransactionUtils.getLastCompletedTxnInstantAndMetadata(metaClient) : Option.empty();
this.pendingInflightAndRequestedInstants = TransactionUtils.getInflightAndRequestedInstants(metaClient);
this.pendingInflightAndRequestedInstants.remove(instantTime);
@@ -519,6 +519,8 @@ public void preWrite(String instantTime, WriteOperationType writeOperationType,
*/
protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option> extraMetadata) {
try {
+ context.setJobStatus(this.getClass().getSimpleName(),"Cleaning up marker directories for commit " + instantTime + " in table "
+ + config.getTableName());
// Delete the marker directory for the instant.
WriteMarkersFactory.get(config.getMarkersType(), table, instantTime)
.quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
@@ -533,7 +535,8 @@ protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, Stri
*/
protected void mayBeCleanAndArchive(HoodieTable table) {
autoCleanOnCommit();
- autoArchiveOnCommit(table);
+ // reload table to that timeline reflects the clean commit
+ autoArchiveOnCommit(createTable(config, hadoopConf));
}
protected void runTableServicesInline(HoodieTable table, HoodieCommitMetadata metadata, Option> extraMetadata) {
@@ -1074,7 +1077,8 @@ protected boolean scheduleCleaningAtInstant(String instantTime, Option scheduleTableService(Option> extraMetadata, TableServiceType tableServiceType) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
@@ -1086,20 +1090,11 @@ public Option scheduleTableService(Option> extraMeta
*
* @param extraMetadata Metadata to pass onto the scheduled service instant
* @param tableServiceType Type of table service to schedule
- * @return
+ *
+ * @return The given instant time option or empty if no table service plan is scheduled
*/
- public Option scheduleTableService(String instantTime, Option> extraMetadata,
- TableServiceType tableServiceType) {
- // A lock is required to guard against race conditions between an on-going writer and scheduling a table service.
- final Option inflightInstant = Option.of(new HoodieInstant(HoodieInstant.State.REQUESTED,
- tableServiceType.getAction(), instantTime));
- try {
- this.txnManager.beginTransaction(inflightInstant, Option.empty());
- LOG.info("Scheduling table service " + tableServiceType);
- return tableServiceClient.scheduleTableServiceInternal(instantTime, extraMetadata, tableServiceType);
- } finally {
- this.txnManager.endTransaction(inflightInstant);
- }
+ public Option scheduleTableService(String instantTime, Option> extraMetadata, TableServiceType tableServiceType) {
+ return tableServiceClient.scheduleTableService(instantTime, extraMetadata, tableServiceType);
}
public HoodieMetrics getMetrics() {
@@ -1277,8 +1272,21 @@ protected void tryUpgrade(HoodieTableMetaClient metaClient, Option insta
}
}
+ /**
+ * Rolls back the failed delta commits corresponding to the indexing action.
+ *
+ * TODO(HUDI-5733): This should be cleaned up once the proper fix of rollbacks
+ * in the metadata table is landed.
+ *
+ * @return {@code true} if rollback happens; {@code false} otherwise.
+ */
+ public boolean lazyRollbackFailedIndexing() {
+ return tableServiceClient.rollbackFailedIndexingCommits();
+ }
+
/**
* Rollback failed writes if any.
+ *
* @return true if rollback happened. false otherwise.
*/
public boolean rollbackFailedWrites() {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/FailOnFirstErrorWriteStatus.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/FailOnFirstErrorWriteStatus.java
new file mode 100644
index 0000000000000..40ccae0f000f1
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/FailOnFirstErrorWriteStatus.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * This class can be used as WriteStatus when we want to fail fast and at the first available exception/error.
+ */
+public class FailOnFirstErrorWriteStatus extends WriteStatus {
+ private static final Logger LOG = LoggerFactory.getLogger(FailOnFirstErrorWriteStatus.class);
+
+ public FailOnFirstErrorWriteStatus(Boolean trackSuccessRecords, Double failureFraction) {
+ super(trackSuccessRecords, failureFraction);
+ }
+
+ @Override
+ public void markFailure(HoodieRecord record, Throwable t, Option> optionalRecordMetadata) {
+ LOG.error(String.format("Error writing record %s with data %s and optionalRecordMetadata %s", record, record.getData(),
+ optionalRecordMetadata.orElse(Collections.emptyMap()), t));
+ throw new HoodieException("Error writing record " + record + ": " + t.getMessage());
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
index ddc5d41448399..56ea142a06068 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
@@ -436,6 +436,8 @@ private Stream getCommitInstantsToArchive() throws IOException {
// The clustering commit instant can not be archived unless we ensure that the replaced files have been cleaned,
// without the replaced files metadata on the timeline, the fs view would expose duplicates for readers.
+ // Meanwhile, when inline or async clustering is enabled, we need to ensure that there is a commit in the active timeline
+ // to check whether the file slice generated in pending clustering after archive isn't committed.
Option oldestInstantToRetainForClustering =
ClusteringUtils.getOldestInstantToRetainForClustering(table.getActiveTimeline(), table.getMetaClient());
@@ -451,7 +453,7 @@ private Stream getCommitInstantsToArchive() throws IOException {
return !(firstSavepoint.isPresent() && compareTimestamps(firstSavepoint.get().getTimestamp(), LESSER_THAN_OR_EQUALS, s.getTimestamp()));
}
}).filter(s -> {
- // Ensure commits >= the oldest pending compaction commit is retained
+ // Ensure commits >= the oldest pending compaction/replace commit is retained
return oldestPendingCompactionAndReplaceInstant
.map(instant -> compareTimestamps(instant.getTimestamp(), GREATER_THAN, s.getTimestamp()))
.orElse(true);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java
index b306d6c540017..54e88fcca225b 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java
@@ -101,7 +101,18 @@ public void markSuccess(HoodieRecord record, Option> optiona
String eventTimeVal = optionalRecordMetadata.get().getOrDefault(METADATA_EVENT_TIME_KEY, null);
try {
if (!StringUtils.isNullOrEmpty(eventTimeVal)) {
- long eventTime = DateTimeUtils.parseDateTime(eventTimeVal).toEpochMilli();
+ int length = eventTimeVal.length();
+ long millisEventTime;
+ // eventTimeVal in seconds unit
+ if (length == 10) {
+ millisEventTime = Long.parseLong(eventTimeVal) * 1000;
+ } else if (length == 13) {
+ // eventTimeVal in millis unit
+ millisEventTime = Long.parseLong(eventTimeVal);
+ } else {
+ throw new IllegalArgumentException("not support event_time format:" + eventTimeVal);
+ }
+ long eventTime = DateTimeUtils.parseDateTime(Long.toString(millisEventTime)).toEpochMilli();
stat.setMinEventTime(eventTime);
stat.setMaxEventTime(eventTime);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
index e07bde915a089..6fcc975004564 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
@@ -95,7 +95,9 @@ public void startServer() throws IOException {
.earlyConflictDetectionCheckCommitConflict(writeConfig.earlyConflictDetectionCheckCommitConflict())
.asyncConflictDetectorInitialDelayMs(writeConfig.getAsyncConflictDetectorInitialDelayMs())
.asyncConflictDetectorPeriodMs(writeConfig.getAsyncConflictDetectorPeriodMs())
- .earlyConflictDetectionMaxAllowableHeartbeatIntervalInMs(writeConfig.getHoodieClientHeartbeatIntervalInMs());
+ .earlyConflictDetectionMaxAllowableHeartbeatIntervalInMs(
+ writeConfig.getHoodieClientHeartbeatIntervalInMs()
+ * writeConfig.getHoodieClientHeartbeatTolerableMisses());
}
server = new TimelineService(context, hadoopConf.newCopy(), timelineServiceConfBuilder.build(),
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/BucketIndexConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/BucketIndexConcurrentFileWritesConflictResolutionStrategy.java
new file mode 100644
index 0000000000000..a15a4cc533c82
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/BucketIndexConcurrentFileWritesConflictResolutionStrategy.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.transaction;
+
+import org.apache.hudi.index.bucket.BucketIdentifier;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class is a basic implementation of a conflict resolution strategy for concurrent writes {@link ConflictResolutionStrategy} using bucket index.
+ */
+public class BucketIndexConcurrentFileWritesConflictResolutionStrategy
+ extends SimpleConcurrentFileWritesConflictResolutionStrategy {
+ private static final Logger LOG = LoggerFactory.getLogger(BucketIndexConcurrentFileWritesConflictResolutionStrategy.class);
+
+ @Override
+ public boolean hasConflict(ConcurrentOperation thisOperation, ConcurrentOperation otherOperation) {
+ // TODO : UUID's can clash even for insert/insert, handle that case.
+ Set partitionBucketIdSetForFirstInstant = thisOperation
+ .getMutatedPartitionAndFileIds()
+ .stream()
+ .map(partitionAndFileId ->
+ BucketIdentifier.partitionBucketIdStr(partitionAndFileId.getLeft(), BucketIdentifier.bucketIdFromFileId(partitionAndFileId.getRight()))
+ ).collect(Collectors.toSet());
+ Set partitionBucketIdSetForSecondInstant = otherOperation
+ .getMutatedPartitionAndFileIds()
+ .stream()
+ .map(partitionAndFileId ->
+ BucketIdentifier.partitionBucketIdStr(partitionAndFileId.getLeft(), BucketIdentifier.bucketIdFromFileId(partitionAndFileId.getRight()))
+ ).collect(Collectors.toSet());
+ Set intersection = new HashSet<>(partitionBucketIdSetForFirstInstant);
+ intersection.retainAll(partitionBucketIdSetForSecondInstant);
+ if (!intersection.isEmpty()) {
+ LOG.info("Found conflicting writes between first operation = " + thisOperation
+ + ", second operation = " + otherOperation + " , intersecting bucket ids " + intersection);
+ return true;
+ }
+ return false;
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
index 4e69b69cf4712..20874849bc30f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
@@ -27,6 +27,7 @@
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
import java.io.IOException;
import java.util.Collections;
@@ -38,7 +39,7 @@
import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.LOG_COMPACTION_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
-import static org.apache.hudi.common.util.CommitUtils.getFileIdWithoutSuffixAndRelativePathsFromSpecificRecord;
+import static org.apache.hudi.common.util.CommitUtils.getPartitionAndFileIdWithoutSuffixFromSpecificRecord;
/**
* This class is used to hold all information used to identify how to resolve conflicts between instants.
@@ -53,7 +54,7 @@ public class ConcurrentOperation {
private final String actionState;
private final String actionType;
private final String instantTime;
- private Set mutatedFileIds = Collections.EMPTY_SET;
+ private Set> mutatedPartitionAndFileIds = Collections.emptySet();
public ConcurrentOperation(HoodieInstant instant, HoodieTableMetaClient metaClient) throws IOException {
this.metadataWrapper = new HoodieMetadataWrapper(MetadataConversionUtils.createMetaWrapper(instant, metaClient));
@@ -89,8 +90,8 @@ public WriteOperationType getOperationType() {
return operationType;
}
- public Set getMutatedFileIds() {
- return mutatedFileIds;
+ public Set> getMutatedPartitionAndFileIds() {
+ return mutatedPartitionAndFileIds;
}
public Option getCommitMetadataOption() {
@@ -102,21 +103,21 @@ private void init(HoodieInstant instant) {
switch (getInstantActionType()) {
case COMPACTION_ACTION:
this.operationType = WriteOperationType.COMPACT;
- this.mutatedFileIds = this.metadataWrapper.getMetadataFromTimeline().getHoodieCompactionPlan().getOperations()
+ this.mutatedPartitionAndFileIds = this.metadataWrapper.getMetadataFromTimeline().getHoodieCompactionPlan().getOperations()
.stream()
- .map(op -> op.getFileId())
+ .map(operation -> Pair.of(operation.getPartitionPath(), operation.getFileId()))
.collect(Collectors.toSet());
break;
case COMMIT_ACTION:
case DELTA_COMMIT_ACTION:
- this.mutatedFileIds = getFileIdWithoutSuffixAndRelativePathsFromSpecificRecord(this.metadataWrapper.getMetadataFromTimeline().getHoodieCommitMetadata()
- .getPartitionToWriteStats()).keySet();
+ this.mutatedPartitionAndFileIds = getPartitionAndFileIdWithoutSuffixFromSpecificRecord(this.metadataWrapper.getMetadataFromTimeline().getHoodieCommitMetadata()
+ .getPartitionToWriteStats());
this.operationType = WriteOperationType.fromValue(this.metadataWrapper.getMetadataFromTimeline().getHoodieCommitMetadata().getOperationType());
break;
case REPLACE_COMMIT_ACTION:
if (instant.isCompleted()) {
- this.mutatedFileIds = getFileIdWithoutSuffixAndRelativePathsFromSpecificRecord(
- this.metadataWrapper.getMetadataFromTimeline().getHoodieReplaceCommitMetadata().getPartitionToWriteStats()).keySet();
+ this.mutatedPartitionAndFileIds = getPartitionAndFileIdWithoutSuffixFromSpecificRecord(
+ this.metadataWrapper.getMetadataFromTimeline().getHoodieReplaceCommitMetadata().getPartitionToWriteStats());
this.operationType = WriteOperationType.fromValue(this.metadataWrapper.getMetadataFromTimeline().getHoodieReplaceCommitMetadata().getOperationType());
} else {
// we need to have different handling for requested and inflight replacecommit because
@@ -127,16 +128,16 @@ private void init(HoodieInstant instant) {
if (instant.isRequested()) {
// for insert_overwrite/insert_overwrite_table clusteringPlan will be empty
if (requestedReplaceMetadata != null && requestedReplaceMetadata.getClusteringPlan() != null) {
- this.mutatedFileIds = getFileIdsFromRequestedReplaceMetadata(requestedReplaceMetadata);
+ this.mutatedPartitionAndFileIds = getPartitionAndFileIdsFromRequestedReplaceMetadata(requestedReplaceMetadata);
this.operationType = WriteOperationType.CLUSTER;
}
} else {
if (inflightCommitMetadata != null) {
- this.mutatedFileIds = getFileIdWithoutSuffixAndRelativePathsFromSpecificRecord(inflightCommitMetadata.getPartitionToWriteStats()).keySet();
+ this.mutatedPartitionAndFileIds = getPartitionAndFileIdWithoutSuffixFromSpecificRecord(inflightCommitMetadata.getPartitionToWriteStats());
this.operationType = WriteOperationType.fromValue(this.metadataWrapper.getMetadataFromTimeline().getHoodieInflightReplaceMetadata().getOperationType());
} else if (requestedReplaceMetadata != null) {
// inflight replacecommit metadata is empty due to clustering, read fileIds from requested replacecommit
- this.mutatedFileIds = getFileIdsFromRequestedReplaceMetadata(requestedReplaceMetadata);
+ this.mutatedPartitionAndFileIds = getPartitionAndFileIdsFromRequestedReplaceMetadata(requestedReplaceMetadata);
this.operationType = WriteOperationType.CLUSTER;
}
// NOTE: it cannot be the case that instant is inflight, and both the requested and inflight replacecommit metadata are empty
@@ -151,7 +152,7 @@ private void init(HoodieInstant instant) {
case COMMIT_ACTION:
case DELTA_COMMIT_ACTION:
case LOG_COMPACTION_ACTION:
- this.mutatedFileIds = CommitUtils.getFileIdWithoutSuffixAndRelativePaths(this.metadataWrapper.getCommitMetadata().getPartitionToWriteStats()).keySet();
+ this.mutatedPartitionAndFileIds = CommitUtils.getPartitionAndFileIdWithoutSuffix(this.metadataWrapper.getCommitMetadata().getPartitionToWriteStats());
this.operationType = this.metadataWrapper.getCommitMetadata().getOperationType();
break;
default:
@@ -160,12 +161,12 @@ private void init(HoodieInstant instant) {
}
}
- private static Set getFileIdsFromRequestedReplaceMetadata(HoodieRequestedReplaceMetadata requestedReplaceMetadata) {
+ private static Set> getPartitionAndFileIdsFromRequestedReplaceMetadata(HoodieRequestedReplaceMetadata requestedReplaceMetadata) {
return requestedReplaceMetadata
.getClusteringPlan().getInputGroups()
.stream()
.flatMap(ig -> ig.getSlices().stream())
- .map(file -> file.getFileId())
+ .map(fileSlice -> Pair.of(fileSlice.getPartitionPath(), fileSlice.getFileId()))
.collect(Collectors.toSet());
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java
index 84fac2db004a8..7ed6d51038c08 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java
@@ -42,13 +42,12 @@ public class DirectMarkerTransactionManager extends TransactionManager {
private final String filePath;
public DirectMarkerTransactionManager(HoodieWriteConfig config, FileSystem fs, String partitionPath, String fileId) {
- super(new LockManager(config, fs, createUpdatedLockProps(config, partitionPath, fileId)),
- config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl());
+ super(new LockManager(config, fs, createUpdatedLockProps(config, partitionPath, fileId)), config.isLockRequired());
this.filePath = partitionPath + "/" + fileId;
}
public void beginTransaction(String newTxnOwnerInstantTime) {
- if (isOptimisticConcurrencyControlEnabled) {
+ if (isLockRequired) {
LOG.info("Transaction starting for " + newTxnOwnerInstantTime + " and " + filePath);
lockManager.lock();
@@ -58,7 +57,7 @@ public void beginTransaction(String newTxnOwnerInstantTime) {
}
public void endTransaction(String currentTxnOwnerInstantTime) {
- if (isOptimisticConcurrencyControlEnabled) {
+ if (isLockRequired) {
LOG.info("Transaction ending with transaction owner " + currentTxnOwnerInstantTime
+ " for " + filePath);
if (reset(Option.of(getInstant(currentTxnOwnerInstantTime)), Option.empty(), Option.empty())) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
index a622486b17d91..22bcd8d0e0cc9 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -25,6 +25,7 @@
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieWriteConflictException;
import org.apache.hudi.table.HoodieTable;
import org.apache.log4j.LogManager;
@@ -72,10 +73,10 @@ public Stream getCandidateInstants(HoodieActiveTimeline activeTim
@Override
public boolean hasConflict(ConcurrentOperation thisOperation, ConcurrentOperation otherOperation) {
// TODO : UUID's can clash even for insert/insert, handle that case.
- Set fileIdsSetForFirstInstant = thisOperation.getMutatedFileIds();
- Set fileIdsSetForSecondInstant = otherOperation.getMutatedFileIds();
- Set intersection = new HashSet<>(fileIdsSetForFirstInstant);
- intersection.retainAll(fileIdsSetForSecondInstant);
+ Set> partitionAndFileIdsSetForFirstInstant = thisOperation.getMutatedPartitionAndFileIds();
+ Set> partitionAndFileIdsSetForSecondInstant = otherOperation.getMutatedPartitionAndFileIds();
+ Set> intersection = new HashSet<>(partitionAndFileIdsSetForFirstInstant);
+ intersection.retainAll(partitionAndFileIdsSetForSecondInstant);
if (!intersection.isEmpty()) {
LOG.info("Found conflicting writes between first operation = " + thisOperation
+ ", second operation = " + otherOperation + " , intersecting file ids " + intersection);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/TransactionManager.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/TransactionManager.java
index 7fddf8a944b06..d1163abac598a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/TransactionManager.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/TransactionManager.java
@@ -37,23 +37,22 @@ public class TransactionManager implements Serializable {
protected static final Logger LOG = LogManager.getLogger(TransactionManager.class);
protected final LockManager lockManager;
- protected final boolean isOptimisticConcurrencyControlEnabled;
+ protected final boolean isLockRequired;
protected Option currentTxnOwnerInstant = Option.empty();
private Option lastCompletedTxnOwnerInstant = Option.empty();
public TransactionManager(HoodieWriteConfig config, FileSystem fs) {
- this(new LockManager(config, fs),
- config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl());
+ this(new LockManager(config, fs), config.isLockRequired());
}
- protected TransactionManager(LockManager lockManager, boolean isOptimisticConcurrencyControlEnabled) {
+ protected TransactionManager(LockManager lockManager, boolean isLockRequired) {
this.lockManager = lockManager;
- this.isOptimisticConcurrencyControlEnabled = isOptimisticConcurrencyControlEnabled;
+ this.isLockRequired = isLockRequired;
}
public void beginTransaction(Option newTxnOwnerInstant,
Option lastCompletedTxnOwnerInstant) {
- if (isOptimisticConcurrencyControlEnabled) {
+ if (isLockRequired) {
LOG.info("Transaction starting for " + newTxnOwnerInstant
+ " with latest completed transaction instant " + lastCompletedTxnOwnerInstant);
lockManager.lock();
@@ -64,7 +63,7 @@ public void beginTransaction(Option newTxnOwnerInstant,
}
public void endTransaction(Option currentTxnOwnerInstant) {
- if (isOptimisticConcurrencyControlEnabled) {
+ if (isLockRequired) {
LOG.info("Transaction ending with transaction owner " + currentTxnOwnerInstant);
if (reset(currentTxnOwnerInstant, Option.empty(), Option.empty())) {
lockManager.unlock();
@@ -85,7 +84,7 @@ protected synchronized boolean reset(Option callerInstant,
}
public void close() {
- if (isOptimisticConcurrencyControlEnabled) {
+ if (isLockRequired) {
lockManager.close();
LOG.info("Transaction manager closed");
}
@@ -103,7 +102,7 @@ public Option getCurrentTransactionOwner() {
return currentTxnOwnerInstant;
}
- public boolean isOptimisticConcurrencyControlEnabled() {
- return isOptimisticConcurrencyControlEnabled;
+ public boolean isLockRequired() {
+ return isLockRequired;
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
index efa644f4b0dce..8956f72560fef 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
@@ -24,8 +24,10 @@
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.config.LockConfiguration;
import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.fs.StorageSchemes;
import org.apache.hudi.common.lock.LockProvider;
import org.apache.hudi.common.lock.LockState;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -41,6 +43,8 @@
import java.io.IOException;
import java.io.Serializable;
import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.apache.hudi.common.config.LockConfiguration.FILESYSTEM_LOCK_EXPIRE_PROP_KEY;
@@ -69,13 +73,17 @@ public FileSystemBasedLockProvider(final LockConfiguration lockConfiguration, fi
String lockDirectory = lockConfiguration.getConfig().getString(FILESYSTEM_LOCK_PATH_PROP_KEY, null);
if (StringUtils.isNullOrEmpty(lockDirectory)) {
lockDirectory = lockConfiguration.getConfig().getString(HoodieWriteConfig.BASE_PATH.key())
- + Path.SEPARATOR + HoodieTableMetaClient.METAFOLDER_NAME;
+ + Path.SEPARATOR + HoodieTableMetaClient.METAFOLDER_NAME;
}
this.lockTimeoutMinutes = lockConfiguration.getConfig().getInteger(FILESYSTEM_LOCK_EXPIRE_PROP_KEY);
this.lockFile = new Path(lockDirectory + Path.SEPARATOR + LOCK_FILE_NAME);
this.lockInfo = new LockInfo();
this.sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
this.fs = FSUtils.getFs(this.lockFile.toString(), configuration);
+ List customSupportedFSs = lockConfiguration.getConfig().getStringList(HoodieCommonConfig.HOODIE_FS_ATOMIC_CREATION_SUPPORT.key(), ",", new ArrayList<>());
+ if (!customSupportedFSs.contains(this.fs.getScheme()) && !StorageSchemes.isAtomicCreationSupported(this.fs.getScheme())) {
+ throw new HoodieLockException("Unsupported scheme :" + this.fs.getScheme() + ", since this fs can not support atomic creation");
+ }
}
@Override
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/InProcessLockProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/InProcessLockProvider.java
index 2a60473c82c8d..904c99019ec49 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/InProcessLockProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/InProcessLockProvider.java
@@ -124,7 +124,6 @@ public void close() {
lock.writeLock().unlock();
}
LOG.info(getLogMessage(LockState.ALREADY_RELEASED));
- LOCK_INSTANCE_PER_BASEPATH.remove(basePath);
}
private String getLogMessage(LockState state) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java
index effcdf091ae2f..3918f2ed74bd2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java
@@ -68,39 +68,37 @@ public LockManager(HoodieWriteConfig writeConfig, FileSystem fs, TypedProperties
}
public void lock() {
- if (writeConfig.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()) {
- LockProvider lockProvider = getLockProvider();
- int retryCount = 0;
- boolean acquired = false;
- while (retryCount <= maxRetries) {
+ LockProvider lockProvider = getLockProvider();
+ int retryCount = 0;
+ boolean acquired = false;
+ while (retryCount <= maxRetries) {
+ try {
+ metrics.startLockApiTimerContext();
+ acquired = lockProvider.tryLock(writeConfig.getLockAcquireWaitTimeoutInMs(), TimeUnit.MILLISECONDS);
+ if (acquired) {
+ metrics.updateLockAcquiredMetric();
+ break;
+ }
+ metrics.updateLockNotAcquiredMetric();
+ LOG.info("Retrying to acquire lock. Current lock owner information : " + lockProvider.getCurrentOwnerLockInfo());
+ Thread.sleep(maxWaitTimeInMs);
+ } catch (HoodieLockException | InterruptedException e) {
+ metrics.updateLockNotAcquiredMetric();
+ if (retryCount >= maxRetries) {
+ throw new HoodieLockException("Unable to acquire lock, lock object " + lockProvider.getLock(), e);
+ }
try {
- metrics.startLockApiTimerContext();
- acquired = lockProvider.tryLock(writeConfig.getLockAcquireWaitTimeoutInMs(), TimeUnit.MILLISECONDS);
- if (acquired) {
- metrics.updateLockAcquiredMetric();
- break;
- }
- metrics.updateLockNotAcquiredMetric();
- LOG.info("Retrying to acquire lock. Current lock owner information : " + lockProvider.getCurrentOwnerLockInfo());
Thread.sleep(maxWaitTimeInMs);
- } catch (HoodieLockException | InterruptedException e) {
- metrics.updateLockNotAcquiredMetric();
- if (retryCount >= maxRetries) {
- throw new HoodieLockException("Unable to acquire lock, lock object " + lockProvider.getLock(), e);
- }
- try {
- Thread.sleep(maxWaitTimeInMs);
- } catch (InterruptedException ex) {
- // ignore InterruptedException here
- }
- } finally {
- retryCount++;
+ } catch (InterruptedException ex) {
+ // ignore InterruptedException here
}
- }
- if (!acquired) {
- throw new HoodieLockException("Unable to acquire lock, lock object " + lockProvider.getLock());
+ } finally {
+ retryCount++;
}
}
+ if (!acquired) {
+ throw new HoodieLockException("Unable to acquire lock, lock object " + lockProvider.getLock());
+ }
}
/**
@@ -108,10 +106,9 @@ public void lock() {
* and tries to call unlock()
*/
public void unlock() {
- if (writeConfig.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()) {
- getLockProvider().unlock();
- metrics.updateLockHeldTimerMetrics();
- }
+ getLockProvider().unlock();
+ metrics.updateLockHeldTimerMetrics();
+ close();
}
public synchronized LockProvider getLockProvider() {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/DeletePartitionUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/DeletePartitionUtils.java
new file mode 100644
index 0000000000000..92c2065457e0a
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/DeletePartitionUtils.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.utils;
+
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
+import org.apache.hudi.exception.HoodieDeletePartitionException;
+import org.apache.hudi.table.HoodieTable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * A utility class for helper functions when performing a delete partition operation.
+ */
+public class DeletePartitionUtils {
+
+ /**
+ * Check if there are any pending table service actions (requested + inflight) on a table affecting the partitions to
+ * be dropped.
+ *
+ * This check is to prevent a drop-partition from proceeding should a partition have a table service action in
+ * the pending stage. If this is allowed to happen, the filegroup that is an input for a table service action, might
+ * also be a candidate for being replaced. As such, when the table service action and drop-partition commits are
+ * committed, there will be two commits replacing a single filegroup.
+ *
+ * For example, a timeline might have an execution order as such:
+ * 000.replacecommit.requested (clustering filegroup_1 + filegroup_2 -> filegroup_3)
+ * 001.replacecommit.requested, 001.replacecommit.inflight, 0001.replacecommit (drop_partition to replace filegroup_1)
+ * 000.replacecommit.inflight (clustering is executed now)
+ * 000.replacecommit (clustering completed)
+ * For an execution order as shown above, 000.replacecommit and 001.replacecommit will both flag filegroup_1 to be replaced.
+ * This will cause downstream duplicate key errors when a map is being constructed.
+ *
+ * @param table Table to perform validation on
+ * @param partitionsToDrop List of partitions to drop
+ */
+ public static void checkForPendingTableServiceActions(HoodieTable table, List partitionsToDrop) {
+ List instantsOfOffendingPendingTableServiceAction = new ArrayList<>();
+ // ensure that there are no pending inflight clustering/compaction operations involving this partition
+ SyncableFileSystemView fileSystemView = (SyncableFileSystemView) table.getSliceView();
+
+ // separating the iteration of pending compaction operations from clustering as they return different stream types
+ Stream.concat(fileSystemView.getPendingCompactionOperations(), fileSystemView.getPendingLogCompactionOperations())
+ .filter(op -> partitionsToDrop.contains(op.getRight().getPartitionPath()))
+ .forEach(op -> instantsOfOffendingPendingTableServiceAction.add(op.getLeft()));
+
+ fileSystemView.getFileGroupsInPendingClustering()
+ .filter(fgIdInstantPair -> partitionsToDrop.contains(fgIdInstantPair.getLeft().getPartitionPath()))
+ .forEach(x -> instantsOfOffendingPendingTableServiceAction.add(x.getRight().getTimestamp()));
+
+ if (instantsOfOffendingPendingTableServiceAction.size() > 0) {
+ throw new HoodieDeletePartitionException("Failed to drop partitions. "
+ + "Please ensure that there are no pending table service actions (clustering/compaction) for the partitions to be deleted: " + partitionsToDrop + ". "
+ + "Instant(s) of offending pending table service action: "
+ + instantsOfOffendingPendingTableServiceAction.stream().distinct().collect(Collectors.toList()));
+ }
+ }
+
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java
index 681ca20baeebf..fdfdab5897afe 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java
@@ -63,7 +63,14 @@ public class HoodieArchivalConfig extends HoodieConfig {
public static final ConfigProperty DELETE_ARCHIVED_INSTANT_PARALLELISM_VALUE = ConfigProperty
.key("hoodie.archive.delete.parallelism")
.defaultValue(100)
- .withDocumentation("Parallelism for deleting archived hoodie commits.");
+ .withDocumentation("When performing archival operation, Hudi needs to delete the files of "
+ + "the archived instants in the active timeline in .hoodie folder. The file deletion "
+ + "also happens after merging small archived files into larger ones if enabled. "
+ + "This config limits the Spark parallelism for deleting files in both cases, i.e., "
+ + "parallelism of deleting files does not go above the configured value and the "
+ + "parallelism is the number of files to delete if smaller than the "
+ + "configured value. If you see that the file deletion in archival operation is slow "
+ + "because of the limited parallelism, you can increase this to tune the performance.");
public static final ConfigProperty MIN_COMMITS_TO_KEEP = ConfigProperty
.key("hoodie.keep.min.commits")
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java
index 0b9116b01c9cc..a6ee8ab41606c 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java
@@ -96,7 +96,15 @@ public class HoodieBootstrapConfig extends HoodieConfig {
.key("hoodie.bootstrap.parallelism")
.defaultValue("1500")
.sinceVersion("0.6.0")
- .withDocumentation("Parallelism value to be used to bootstrap data into hudi");
+ .withDocumentation("For metadata-only bootstrap, Hudi parallelizes the operation so that "
+ + "each table partition is handled by one Spark task. This config limits the number "
+ + "of parallelism. We pick the configured parallelism if the number of table partitions "
+ + "is larger than this configured value. The parallelism is assigned to the number of "
+ + "table partitions if it is smaller than the configured value. For full-record "
+ + "bootstrap, i.e., BULK_INSERT operation of the records, this configured value is "
+ + "passed as the BULK_INSERT shuffle parallelism (`hoodie.bulkinsert.shuffle.parallelism`), "
+ + "determining the BULK_INSERT write behavior. If you see that the bootstrap is slow "
+ + "due to the limited parallelism, you can increase this.");
public static final ConfigProperty PARTITION_SELECTOR_REGEX_PATTERN = ConfigProperty
.key("hoodie.bootstrap.mode.selector.regex")
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCleanConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCleanConfig.java
index 30289e1acbab4..c1b66a371d19c 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCleanConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCleanConfig.java
@@ -29,6 +29,7 @@
import org.apache.hudi.table.action.clean.CleaningTriggerStrategy;
import javax.annotation.concurrent.Immutable;
+
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
@@ -120,7 +121,17 @@ public class HoodieCleanConfig extends HoodieConfig {
public static final ConfigProperty CLEANER_PARALLELISM_VALUE = ConfigProperty
.key("hoodie.cleaner.parallelism")
.defaultValue("200")
- .withDocumentation("Parallelism for the cleaning operation. Increase this if cleaning becomes slow.");
+ .withDocumentation("This config controls the behavior of both the cleaning plan and "
+ + "cleaning execution. Deriving the cleaning plan is parallelized at the table "
+ + "partition level, i.e., each table partition is processed by one Spark task to figure "
+ + "out the files to clean. The cleaner picks the configured parallelism if the number "
+ + "of table partitions is larger than this configured value. The parallelism is "
+ + "assigned to the number of table partitions if it is smaller than the configured value. "
+ + "The clean execution, i.e., the file deletion, is parallelized at file level, which "
+ + "is the unit of Spark task distribution. Similarly, the actual parallelism cannot "
+ + "exceed the configured value if the number of files is larger. If cleaning plan or "
+ + "execution is slow due to limited parallelism, you can increase this to tune the "
+ + "performance..");
public static final ConfigProperty ALLOW_MULTIPLE_CLEANS = ConfigProperty
.key("hoodie.clean.allow.multiple")
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java
index e22bf1e43d136..e37ff3c46bf05 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java
@@ -22,6 +22,7 @@
import org.apache.hudi.common.config.ConfigGroups;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.table.action.compact.CompactionTriggerStrategy;
import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
import org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy;
@@ -188,11 +189,11 @@ public class HoodieCompactionConfig extends HoodieConfig {
.withDocumentation("Log compaction can be scheduled if the no. of log blocks crosses this threshold value. "
+ "This is effective only when log compaction is enabled via " + INLINE_LOG_COMPACT.key());
- public static final ConfigProperty USE_LOG_RECORD_READER_SCAN_V2 = ConfigProperty
- .key("hoodie.log.record.reader.use.scanV2")
+ public static final ConfigProperty ENABLE_OPTIMIZED_LOG_BLOCKS_SCAN = ConfigProperty
+ .key("hoodie" + HoodieMetadataConfig.OPTIMIZED_LOG_BLOCKS_SCAN)
.defaultValue("false")
.sinceVersion("0.13.0")
- .withDocumentation("ScanV2 logic address all the multiwriter challenges while appending to log files. "
+ .withDocumentation("New optimized scan for log blocks that handles all multi-writer use-cases while appending to log files. "
+ "It also differentiates original blocks written by ingestion writers and compacted blocks written log compaction.");
/** @deprecated Use {@link #INLINE_COMPACT} and its methods instead */
@@ -432,8 +433,8 @@ public Builder withLogCompactionBlocksThreshold(String logCompactionBlocksThresh
return this;
}
- public Builder withLogRecordReaderScanV2(String useLogRecordReaderScanV2) {
- compactionConfig.setValue(USE_LOG_RECORD_READER_SCAN_V2, useLogRecordReaderScanV2);
+ public Builder withEnableOptimizedLogBlocksScan(String enableOptimizedLogBlocksScan) {
+ compactionConfig.setValue(ENABLE_OPTIMIZED_LOG_BLOCKS_SCAN, enableOptimizedLogBlocksScan);
return this;
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java
index d52c407028a21..359b9ab6a5289 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java
@@ -31,6 +31,7 @@
@ConfigClassProperty(name = "HBase Index Configs",
groupName = ConfigGroups.Names.WRITE_CLIENT,
+ subGroupName = ConfigGroups.SubGroupNames.INDEX,
description = "Configurations that control indexing behavior "
+ "(when HBase based indexing is enabled), which tags incoming "
+ "records as either inserts or updates to older records.")
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
index 87e9ae576dbee..3152ec34942c2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
@@ -59,10 +59,11 @@
* Indexing related config.
*/
@Immutable
-@ConfigClassProperty(name = "Index Configs",
+@ConfigClassProperty(name = "Common Index Configs",
groupName = ConfigGroups.Names.WRITE_CLIENT,
- description = "Configurations that control indexing behavior, "
- + "which tags incoming records as either inserts or updates to older records.")
+ subGroupName = ConfigGroups.SubGroupNames.INDEX,
+ areCommonConfigs = true,
+ description = "")
public class HoodieIndexConfig extends HoodieConfig {
private static final Logger LOG = LogManager.getLogger(HoodieIndexConfig.class);
@@ -113,7 +114,10 @@ public class HoodieIndexConfig extends HoodieConfig {
.defaultValue("0")
.withDocumentation("Only applies if index type is BLOOM. "
+ "This is the amount of parallelism for index lookup, which involves a shuffle. "
- + "By default, this is auto computed based on input workload characteristics.");
+ + "By default, this is auto computed based on input workload characteristics. "
+ + "If the parallelism is explicitly configured by the user, the user-configured "
+ + "value is used in defining the actual parallelism. If the indexing stage is slow "
+ + "due to the limited parallelism, you can increase this to tune the performance.");
public static final ConfigProperty BLOOM_INDEX_PRUNE_BY_RANGES = ConfigProperty
.key("hoodie.bloom.index.prune.by.ranges")
@@ -179,13 +183,21 @@ public class HoodieIndexConfig extends HoodieConfig {
.key("hoodie.simple.index.parallelism")
.defaultValue("100")
.withDocumentation("Only applies if index type is SIMPLE. "
- + "This is the amount of parallelism for index lookup, which involves a Spark Shuffle");
+ + "This limits the parallelism of fetching records from the base files of affected "
+ + "partitions. The index picks the configured parallelism if the number of base "
+ + "files is larger than this configured value; otherwise, the number of base files "
+ + "is used as the parallelism. If the indexing stage is slow due to the limited "
+ + "parallelism, you can increase this to tune the performance.");
public static final ConfigProperty GLOBAL_SIMPLE_INDEX_PARALLELISM = ConfigProperty
.key("hoodie.global.simple.index.parallelism")
.defaultValue("100")
.withDocumentation("Only applies if index type is GLOBAL_SIMPLE. "
- + "This is the amount of parallelism for index lookup, which involves a Spark Shuffle");
+ + "This limits the parallelism of fetching records from the base files of all table "
+ + "partitions. The index picks the configured parallelism if the number of base "
+ + "files is larger than this configured value; otherwise, the number of base files "
+ + "is used as the parallelism. If the indexing stage is slow due to the limited "
+ + "parallelism, you can increase this to tune the performance.");
// 1B bloom filter checks happen in 250 seconds. 500ms to read a bloom filter.
// 10M checks in 2500ms, thus amortizing the cost of reading bloom filter across partitions.
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLockConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLockConfig.java
index 2c66ac8e68f91..62aa66942756e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLockConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLockConfig.java
@@ -17,6 +17,7 @@
package org.apache.hudi.config;
+import org.apache.hudi.client.transaction.BucketIndexConcurrentFileWritesConflictResolutionStrategy;
import org.apache.hudi.client.transaction.ConflictResolutionStrategy;
import org.apache.hudi.client.transaction.SimpleConcurrentFileWritesConflictResolutionStrategy;
import org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider;
@@ -26,6 +27,7 @@
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.lock.LockProvider;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.index.HoodieIndex;
import java.io.File;
import java.io.FileReader;
@@ -58,11 +60,11 @@
/**
* Hoodie Configs for Locks.
*/
-@ConfigClassProperty(name = "Locks Configurations",
+@ConfigClassProperty(name = "Common Lock Configurations",
groupName = ConfigGroups.Names.WRITE_CLIENT,
- description = "Configs that control locking mechanisms required for concurrency control "
- + " between writers to a Hudi table. Concurrency between Hudi's own table services "
- + " are auto managed internally.")
+ subGroupName = ConfigGroups.SubGroupNames.LOCK,
+ areCommonConfigs = true,
+ description = "")
public class HoodieLockConfig extends HoodieConfig {
public static final ConfigProperty LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS = ConfigProperty
@@ -81,7 +83,7 @@ public class HoodieLockConfig extends HoodieConfig {
public static final ConfigProperty LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS = ConfigProperty
.key(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY)
- .defaultValue(String.valueOf(2000L))
+ .defaultValue(String.valueOf(5000L))
.sinceVersion("0.8.0")
.withDocumentation("Amount of time to wait between retries on the lock provider by the lock manager");
@@ -185,6 +187,13 @@ public class HoodieLockConfig extends HoodieConfig {
public static final ConfigProperty WRITE_CONFLICT_RESOLUTION_STRATEGY_CLASS_NAME = ConfigProperty
.key(LOCK_PREFIX + "conflict.resolution.strategy")
.defaultValue(SimpleConcurrentFileWritesConflictResolutionStrategy.class.getName())
+ .withInferFunction(hoodieConfig -> {
+ if (HoodieIndex.IndexType.BUCKET.name().equalsIgnoreCase(hoodieConfig.getStringOrDefault(HoodieIndexConfig.INDEX_TYPE, null))) {
+ return Option.of(BucketIndexConcurrentFileWritesConflictResolutionStrategy.class.getName());
+ } else {
+ return Option.of(SimpleConcurrentFileWritesConflictResolutionStrategy.class.getName());
+ }
+ })
.sinceVersion("0.8.0")
.withDocumentation("Lock provider class name, this should be subclass of "
+ "org.apache.hudi.client.transaction.ConflictResolutionStrategy");
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java
index 09c2b09be0321..8be57c05d1c4e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java
@@ -32,8 +32,9 @@
*/
@ConfigClassProperty(name = "Write commit callback configs",
groupName = ConfigGroups.Names.WRITE_CLIENT,
- description = "Controls callback behavior into HTTP endpoints, to push "
- + " notifications on commits on hudi tables.")
+ subGroupName = ConfigGroups.SubGroupNames.COMMIT_CALLBACK,
+ areCommonConfigs = true,
+ description = "")
public class HoodieWriteCommitCallbackConfig extends HoodieConfig {
public static final String CALLBACK_PREFIX = "hoodie.write.commit.callback.";
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 2f36aa725e338..f61ba16a33c6c 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
@@ -53,7 +53,8 @@
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.StringUtils;
-import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.queue.DisruptorWaitStrategyType;
import org.apache.hudi.common.util.queue.ExecutorType;
import org.apache.hudi.config.metrics.HoodieMetricsCloudWatchConfig;
import org.apache.hudi.config.metrics.HoodieMetricsConfig;
@@ -67,6 +68,7 @@
import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.keygen.constant.KeyGeneratorType;
+import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.metrics.MetricsReporterType;
import org.apache.hudi.metrics.datadog.DatadogHttpClient.ApiSite;
import org.apache.hudi.table.RandomFileIdPrefixProvider;
@@ -97,8 +99,8 @@
import java.util.function.Supplier;
import java.util.stream.Collectors;
-import static org.apache.hudi.common.util.queue.ExecutorType.BOUNDED_IN_MEMORY;
-import static org.apache.hudi.common.util.queue.ExecutorType.DISRUPTOR;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+import static org.apache.hudi.common.util.queue.ExecutorType.SIMPLE;
import static org.apache.hudi.config.HoodieCleanConfig.CLEANER_POLICY;
import static org.apache.hudi.table.marker.ConflictDetectionUtils.getDefaultEarlyConflictDetectionStrategy;
@@ -158,23 +160,25 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Key generator class, that implements `org.apache.hudi.keygen.KeyGenerator` "
+ "extract a key out of incoming records.");
- public static final ConfigProperty EXECUTOR_TYPE = ConfigProperty
+ public static final ConfigProperty WRITE_EXECUTOR_TYPE = ConfigProperty
.key("hoodie.write.executor.type")
- .defaultValue(BOUNDED_IN_MEMORY.name())
- .withValidValues(BOUNDED_IN_MEMORY.name(), DISRUPTOR.name())
+ .defaultValue(SIMPLE.name())
+ .withValidValues(Arrays.stream(ExecutorType.values()).map(Enum::name).toArray(String[]::new))
.sinceVersion("0.13.0")
.withDocumentation("Set executor which orchestrates concurrent producers and consumers communicating through a message queue."
- + "BOUNDED_IN_MEMORY(default): Use LinkedBlockingQueue as a bounded in-memory queue, this queue will use extra lock to balance producers and consumer"
+ + "BOUNDED_IN_MEMORY: Use LinkedBlockingQueue as a bounded in-memory queue, this queue will use extra lock to balance producers and consumer"
+ "DISRUPTOR: Use disruptor which a lock free message queue as inner message, this queue may gain better writing performance if lock was the bottleneck. "
- + "SIMPLE: Executor with no inner message queue and no inner lock. Consuming and writing records from iterator directly. Compared with BIM and DISRUPTOR, "
+ + "SIMPLE(default): Executor with no inner message queue and no inner lock. Consuming and writing records from iterator directly. Compared with BIM and DISRUPTOR, "
+ "this queue has no need for additional memory and cpu resources due to lock or multithreading, but also lost some benefits such as speed limit. "
- + "Although DISRUPTOR_EXECUTOR and SIMPLE are still in experimental.");
+ + "Although DISRUPTOR is still experimental.");
public static final ConfigProperty KEYGENERATOR_TYPE = ConfigProperty
.key("hoodie.datasource.write.keygenerator.type")
.defaultValue(KeyGeneratorType.SIMPLE.name())
.withDocumentation("Easily configure one the built-in key generators, instead of specifying the key generator class."
- + "Currently supports SIMPLE, COMPLEX, TIMESTAMP, CUSTOM, NON_PARTITION, GLOBAL_DELETE");
+ + "Currently supports SIMPLE, COMPLEX, TIMESTAMP, CUSTOM, NON_PARTITION, GLOBAL_DELETE. "
+ + "**Note** This is being actively worked on. Please use "
+ + "`hoodie.datasource.write.keygenerator.class` instead.");
public static final ConfigProperty ROLLBACK_USING_MARKERS_ENABLE = ConfigProperty
.key("hoodie.rollback.using.markers")
@@ -223,19 +227,44 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty AVRO_SCHEMA_VALIDATE_ENABLE = ConfigProperty
.key("hoodie.avro.schema.validate")
- .defaultValue("true")
+ .defaultValue("false")
.withDocumentation("Validate the schema used for the write against the latest schema, for backwards compatibility.");
+ public static final ConfigProperty SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP = ConfigProperty
+ .key("hoodie.datasource.write.schema.allow.auto.evolution.column.drop")
+ .defaultValue("false")
+ .sinceVersion("0.13.0")
+ .withDocumentation("Controls whether table's schema is allowed to automatically evolve when "
+ + "incoming batch's schema can have any of the columns dropped. By default, Hudi will not "
+ + "allow this kind of (auto) schema evolution. Set this config to true to allow table's "
+ + "schema to be updated automatically when columns are dropped from the new incoming batch.");
+
public static final ConfigProperty INSERT_PARALLELISM_VALUE = ConfigProperty
.key("hoodie.insert.shuffle.parallelism")
.defaultValue("0")
- .withDocumentation("Parallelism for inserting records into the table. Inserts can shuffle data before writing to tune file sizes and optimize the storage layout.");
+ .withDocumentation("Parallelism for inserting records into the table. Inserts can shuffle "
+ + "data before writing to tune file sizes and optimize the storage layout. Before "
+ + "0.13.0 release, if users do not configure it, Hudi would use 200 as the default "
+ + "shuffle parallelism. From 0.13.0 onwards Hudi by default automatically uses the "
+ + "parallelism deduced by Spark based on the source data. If the shuffle parallelism "
+ + "is explicitly configured by the user, the user-configured parallelism is "
+ + "used in defining the actual parallelism. If you observe small files from the insert "
+ + "operation, we suggest configuring this shuffle parallelism explicitly, so that the "
+ + "parallelism is around total_input_data_size/120MB.");
public static final ConfigProperty BULKINSERT_PARALLELISM_VALUE = ConfigProperty
.key("hoodie.bulkinsert.shuffle.parallelism")
.defaultValue("0")
- .withDocumentation("For large initial imports using bulk_insert operation, controls the parallelism to use for sort modes or custom partitioning done"
- + "before writing records to the table.");
+ .withDocumentation("For large initial imports using bulk_insert operation, controls the "
+ + "parallelism to use for sort modes or custom partitioning done before writing records "
+ + "to the table. Before 0.13.0 release, if users do not configure it, Hudi would use "
+ + "200 as the default shuffle parallelism. From 0.13.0 onwards Hudi by default "
+ + "automatically uses the parallelism deduced by Spark based on the source data or "
+ + "the parallelism based on the logical plan for row writer. If the shuffle parallelism "
+ + "is explicitly configured by the user, the user-configured parallelism is "
+ + "used in defining the actual parallelism. If you observe small files from the bulk insert "
+ + "operation, we suggest configuring this shuffle parallelism explicitly, so that the "
+ + "parallelism is around total_input_data_size/120MB.");
public static final ConfigProperty BULKINSERT_USER_DEFINED_PARTITIONER_SORT_COLUMNS = ConfigProperty
.key("hoodie.bulkinsert.user.defined.partitioner.sort.columns")
@@ -253,33 +282,51 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty UPSERT_PARALLELISM_VALUE = ConfigProperty
.key("hoodie.upsert.shuffle.parallelism")
.defaultValue("0")
- .withDocumentation("Parallelism to use for upsert operation on the table. Upserts can shuffle data to perform index lookups, file sizing, bin packing records optimally"
- + "into file groups.");
+ .withDocumentation("Parallelism to use for upsert operation on the table. Upserts can "
+ + "shuffle data to perform index lookups, file sizing, bin packing records optimally "
+ + "into file groups. Before 0.13.0 release, "
+ + "if users do not configure it, Hudi would use 200 as the default "
+ + "shuffle parallelism. From 0.13.0 onwards Hudi by default automatically uses the "
+ + "parallelism deduced by Spark based on the source data. If the shuffle parallelism "
+ + "is explicitly configured by the user, the user-configured parallelism is "
+ + "used in defining the actual parallelism. If you observe small files from the upsert "
+ + "operation, we suggest configuring this shuffle parallelism explicitly, so that the "
+ + "parallelism is around total_input_data_size/120MB.");
public static final ConfigProperty DELETE_PARALLELISM_VALUE = ConfigProperty
.key("hoodie.delete.shuffle.parallelism")
.defaultValue("0")
- .withDocumentation("Parallelism used for “delete” operation. Delete operations also performs shuffles, similar to upsert operation.");
+ .withDocumentation("Parallelism used for delete operation. Delete operations also performs "
+ + "shuffles, similar to upsert operation. Before 0.13.0 release, "
+ + "if users do not configure it, Hudi would use 200 as the default "
+ + "shuffle parallelism. From 0.13.0 onwards Hudi by default automatically uses the "
+ + "parallelism deduced by Spark based on the source data. If the shuffle parallelism "
+ + "is explicitly configured by the user, the user-configured parallelism is "
+ + "used in defining the actual parallelism.");
public static final ConfigProperty ROLLBACK_PARALLELISM_VALUE = ConfigProperty
.key("hoodie.rollback.parallelism")
.defaultValue("100")
- .withDocumentation("Parallelism for rollback of commits. Rollbacks perform delete of files or logging delete blocks to file groups on storage in parallel.");
+ .withDocumentation("This config controls the parallelism for rollback of commits. "
+ + "Rollbacks perform deletion of files or logging delete blocks to file groups on "
+ + "storage in parallel. The configure value limits the parallelism so that the number "
+ + "of Spark tasks do not exceed the value. If rollback is slow due to the limited "
+ + "parallelism, you can increase this to tune the performance.");
public static final ConfigProperty WRITE_BUFFER_LIMIT_BYTES_VALUE = ConfigProperty
.key("hoodie.write.buffer.limit.bytes")
.defaultValue(String.valueOf(4 * 1024 * 1024))
.withDocumentation("Size of in-memory buffer used for parallelizing network reads and lake storage writes.");
- public static final ConfigProperty WRITE_DISRUPTOR_BUFFER_SIZE = ConfigProperty
- .key("hoodie.write.executor.disruptor.buffer.size")
+ public static final ConfigProperty WRITE_EXECUTOR_DISRUPTOR_BUFFER_LIMIT_BYTES = ConfigProperty
+ .key("hoodie.write.executor.disruptor.buffer.limit.bytes")
.defaultValue(String.valueOf(1024))
.sinceVersion("0.13.0")
.withDocumentation("The size of the Disruptor Executor ring buffer, must be power of 2");
- public static final ConfigProperty WRITE_WAIT_STRATEGY = ConfigProperty
+ public static final ConfigProperty WRITE_EXECUTOR_DISRUPTOR_WAIT_STRATEGY = ConfigProperty
.key("hoodie.write.executor.disruptor.wait.strategy")
- .defaultValue("BLOCKING_WAIT")
+ .defaultValue(DisruptorWaitStrategyType.BLOCKING_WAIT.name())
.sinceVersion("0.13.0")
.withDocumentation("Strategy employed for making Disruptor Executor wait on a cursor. Other options are "
+ "SLEEPING_WAIT, it attempts to be conservative with CPU usage by using a simple busy wait loop"
@@ -581,7 +628,7 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty ASYNC_CONFLICT_DETECTOR_INITIAL_DELAY_MS = ConfigProperty
.key(CONCURRENCY_PREFIX + "async.conflict.detector.initial_delay_ms")
- .defaultValue(30000L)
+ .defaultValue(0L)
.sinceVersion("0.13.0")
.withDocumentation("Used for timeline-server-based markers with "
+ "`AsyncTimelineServerBasedDetectionStrategy`. "
@@ -602,6 +649,12 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("Whether to enable commit conflict checking or not during early "
+ "conflict detection.");
+ public static final ConfigProperty SENSITIVE_CONFIG_KEYS_FILTER = ConfigProperty
+ .key("hoodie.sensitive.config.keys")
+ .defaultValue("ssl,tls,sasl,auth,credentials")
+ .withDocumentation("Comma separated list of filters for sensitive config keys. Delta Streamer "
+ + "will not print any configuration which contains the configured filter. For example with "
+ + "a configured filter `ssl`, value for config `ssl.trustore.location` would be masked.");
private ConsistencyGuardConfig consistencyGuardConfig;
private FileSystemRetryConfig fileSystemRetryConfig;
@@ -1024,7 +1077,7 @@ public String getBasePath() {
}
public HoodieRecordMerger getRecordMerger() {
- List mergers = getSplitStringsOrDefault(RECORD_MERGER_IMPLS).stream()
+ List mergers = StringUtils.split(getStringOrDefault(RECORD_MERGER_IMPLS), ",").stream()
.map(String::trim)
.distinct()
.collect(Collectors.toList());
@@ -1085,6 +1138,10 @@ public boolean shouldValidateAvroSchema() {
return getBoolean(AVRO_SCHEMA_VALIDATE_ENABLE);
}
+ public boolean shouldAllowAutoEvolutionColumnDrop() {
+ return getBooleanOrDefault(SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP);
+ }
+
public String getTableName() {
return getString(TBL_NAME);
}
@@ -1107,7 +1164,7 @@ public String getKeyGeneratorClass() {
}
public ExecutorType getExecutorType() {
- return ExecutorType.valueOf(getStringOrDefault(EXECUTOR_TYPE).toUpperCase(Locale.ROOT));
+ return ExecutorType.valueOf(getStringOrDefault(WRITE_EXECUTOR_TYPE).toUpperCase(Locale.ROOT));
}
public boolean isCDCEnabled() {
@@ -1175,12 +1232,12 @@ public int getWriteBufferLimitBytes() {
return Integer.parseInt(getStringOrDefault(WRITE_BUFFER_LIMIT_BYTES_VALUE));
}
- public Option getWriteExecutorWaitStrategy() {
- return Option.of(getString(WRITE_WAIT_STRATEGY));
+ public String getWriteExecutorDisruptorWaitStrategy() {
+ return getStringOrDefault(WRITE_EXECUTOR_DISRUPTOR_WAIT_STRATEGY);
}
- public Option getDisruptorWriteBufferSize() {
- return Option.of(Integer.parseInt(getStringOrDefault(WRITE_DISRUPTOR_BUFFER_SIZE)));
+ public Integer getWriteExecutorDisruptorWriteBufferLimitBytes() {
+ return Integer.parseInt(getStringOrDefault(WRITE_EXECUTOR_DISRUPTOR_BUFFER_LIMIT_BYTES));
}
public boolean shouldCombineBeforeInsert() {
@@ -1305,8 +1362,8 @@ public int getLogCompactionBlocksThreshold() {
return getInt(HoodieCompactionConfig.LOG_COMPACTION_BLOCKS_THRESHOLD);
}
- public boolean useScanV2ForLogRecordReader() {
- return getBoolean(HoodieCompactionConfig.USE_LOG_RECORD_READER_SCAN_V2);
+ public boolean enableOptimizedLogBlocksScan() {
+ return getBoolean(HoodieCompactionConfig.ENABLE_OPTIMIZED_LOG_BLOCKS_SCAN);
}
public HoodieCleaningPolicy getCleanerPolicy() {
@@ -1862,8 +1919,8 @@ public int getParquetPageSize() {
return getInt(HoodieStorageConfig.PARQUET_PAGE_SIZE);
}
- public int getLogFileDataBlockMaxSize() {
- return getInt(HoodieStorageConfig.LOGFILE_DATA_BLOCK_MAX_SIZE);
+ public long getLogFileDataBlockMaxSize() {
+ return getLong(HoodieStorageConfig.LOGFILE_DATA_BLOCK_MAX_SIZE);
}
public double getParquetCompressionRatio() {
@@ -1987,7 +2044,7 @@ public ApiSite getDatadogApiSite() {
public String getDatadogApiKey() {
if (props.containsKey(HoodieMetricsDatadogConfig.API_KEY.key())) {
return getString(HoodieMetricsDatadogConfig.API_KEY);
-
+
} else {
Supplier apiKeySupplier = ReflectionUtils.loadClass(
getString(HoodieMetricsDatadogConfig.API_KEY_SUPPLIER));
@@ -2194,7 +2251,7 @@ public Integer getHoodieClientHeartbeatTolerableMisses() {
* File listing metadata configs.
*/
public boolean isMetadataTableEnabled() {
- return metadataConfig.enabled();
+ return getBooleanOrDefault(HoodieMetadataConfig.ENABLE);
}
public int getMetadataInsertParallelism() {
@@ -2345,6 +2402,20 @@ public boolean areReleaseResourceEnabled() {
return getBooleanOrDefault(RELEASE_RESOURCE_ENABLE);
}
+ /**
+ * Returns whether the explicit guard of lock is required.
+ */
+ public boolean isLockRequired() {
+ return !isDefaultLockProvider() || getWriteConcurrencyMode().supportsOptimisticConcurrencyControl();
+ }
+
+ /**
+ * Returns whether the lock provider is default.
+ */
+ private boolean isDefaultLockProvider() {
+ return HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.defaultValue().equals(getLockProviderClass());
+ }
+
/**
* Layout configs.
*/
@@ -2363,7 +2434,7 @@ public boolean isMetaserverEnabled() {
* CDC supplemental logging mode.
*/
public HoodieCDCSupplementalLoggingMode getCDCSupplementalLoggingMode() {
- return HoodieCDCSupplementalLoggingMode.parse(
+ return HoodieCDCSupplementalLoggingMode.valueOf(
getStringOrDefault(HoodieTableConfig.CDC_SUPPLEMENTAL_LOGGING_MODE));
}
@@ -2450,6 +2521,11 @@ public Builder withAvroSchemaValidate(boolean enable) {
return this;
}
+ public Builder withAllowAutoEvolutionColumnDrop(boolean shouldAllowDroppedColumns) {
+ writeConfig.setValue(SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP, String.valueOf(shouldAllowDroppedColumns));
+ return this;
+ }
+
public Builder forTable(String tableName) {
writeConfig.setValue(TBL_NAME, tableName);
return this;
@@ -2481,7 +2557,7 @@ public Builder withKeyGenerator(String keyGeneratorClass) {
}
public Builder withExecutorType(String executorClass) {
- writeConfig.setValue(EXECUTOR_TYPE, executorClass);
+ writeConfig.setValue(WRITE_EXECUTOR_TYPE, executorClass);
return this;
}
@@ -2536,13 +2612,13 @@ public Builder withWriteBufferLimitBytes(int writeBufferLimit) {
return this;
}
- public Builder withWriteWaitStrategy(String waitStrategy) {
- writeConfig.setValue(WRITE_WAIT_STRATEGY, String.valueOf(waitStrategy));
+ public Builder withWriteExecutorDisruptorWaitStrategy(String waitStrategy) {
+ writeConfig.setValue(WRITE_EXECUTOR_DISRUPTOR_WAIT_STRATEGY, String.valueOf(waitStrategy));
return this;
}
- public Builder withWriteBufferSize(int size) {
- writeConfig.setValue(WRITE_DISRUPTOR_BUFFER_SIZE, String.valueOf(size));
+ public Builder withWriteExecutorDisruptorWriteBufferLimitBytes(long size) {
+ writeConfig.setValue(WRITE_EXECUTOR_DISRUPTOR_BUFFER_LIMIT_BYTES, String.valueOf(size));
return this;
}
@@ -2881,41 +2957,34 @@ protected void setDefaults() {
// isLockProviderPropertySet must be fetched before setting defaults of HoodieLockConfig
final TypedProperties writeConfigProperties = writeConfig.getProps();
- final boolean isLockProviderPropertySet = writeConfigProperties.containsKey(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME)
- || writeConfigProperties.containsKey(HoodieLockConfig.LOCK_PROVIDER_CLASS_PROP);
+ final boolean isLockProviderPropertySet = writeConfigProperties.containsKey(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key());
writeConfig.setDefaultOnCondition(!isLockConfigSet,
HoodieLockConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
autoAdjustConfigsForConcurrencyMode(isLockProviderPropertySet);
}
+ private boolean isLockRequiredForSingleWriter() {
+ // When metadata table is enabled, lock provider must be used for
+ // single writer with async table services.
+ // Async table services can update the metadata table and a lock provider is
+ // needed to guard against any concurrent table write operations. If user has
+ // not configured any lock provider, let's use the InProcess lock provider.
+ return writeConfig.isMetadataTableEnabled() && writeConfig.areAnyTableServicesAsync()
+ && !writeConfig.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl();
+ }
+
private void autoAdjustConfigsForConcurrencyMode(boolean isLockProviderPropertySet) {
- if (writeConfig.isAutoAdjustLockConfigs()) {
+ if (!isLockProviderPropertySet && writeConfig.isAutoAdjustLockConfigs() && isLockRequiredForSingleWriter()) {
// auto adjustment is required only for deltastreamer and spark streaming where async table services can be executed in the same JVM.
- boolean isMetadataTableEnabled = writeConfig.getBoolean(HoodieMetadataConfig.ENABLE);
-
- if (isMetadataTableEnabled) {
- // When metadata table is enabled, optimistic concurrency control must be used for
- // single writer with async table services.
- // Async table services can update the metadata table and a lock provider is
- // needed to guard against any concurrent table write operations. If user has
- // not configured any lock provider, let's use the InProcess lock provider.
- boolean areTableServicesEnabled = writeConfig.areTableServicesEnabled();
- boolean areAsyncTableServicesEnabled = writeConfig.areAnyTableServicesAsync();
- if (!isLockProviderPropertySet && areTableServicesEnabled && areAsyncTableServicesEnabled) {
- // This is targeted at Single writer with async table services
- // If user does not set the lock provider, likely that the concurrency mode is not set either
- // Override the configs for metadata table
- writeConfig.setValue(WRITE_CONCURRENCY_MODE.key(),
- WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.value());
- writeConfig.setValue(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key(),
- InProcessLockProvider.class.getName());
- LOG.info(String.format("Automatically set %s=%s and %s=%s since user has not set the "
- + "lock provider for single writer with async table services",
- WRITE_CONCURRENCY_MODE.key(), WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.value(),
- HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key(), InProcessLockProvider.class.getName()));
- }
- }
+ // This is targeted at Single writer with async table services
+ // If user does not set the lock provider, likely that the concurrency mode is not set either
+ // Override the configs for metadata table
+ writeConfig.setValue(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key(),
+ InProcessLockProvider.class.getName());
+ LOG.info(String.format("Automatically set %s=%s since user has not set the "
+ + "lock provider for single writer with async table services",
+ HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key(), InProcessLockProvider.class.getName()));
}
// We check if "hoodie.cleaner.policy.failed.writes"
@@ -2937,9 +3006,14 @@ private void validate() {
// Ensure Layout Version is good
new TimelineLayoutVersion(Integer.parseInt(layoutVersion));
Objects.requireNonNull(writeConfig.getString(BASE_PATH));
+ if (writeConfig.isEarlyConflictDetectionEnable()) {
+ checkArgument(writeConfig.getString(WRITE_CONCURRENCY_MODE)
+ .equalsIgnoreCase(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.value()),
+ "To use early conflict detection, set hoodie.write.concurrency.mode=OPTIMISTIC_CONCURRENCY_CONTROL");
+ }
if (writeConfig.getString(WRITE_CONCURRENCY_MODE)
.equalsIgnoreCase(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.value())) {
- ValidationUtils.checkArgument(!writeConfig.getString(HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY)
+ checkArgument(!writeConfig.getString(HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY)
.equals(HoodieFailedWritesCleaningPolicy.EAGER.name()), "To enable optimistic concurrency control, set hoodie.cleaner.policy.failed.writes=LAZY");
}
@@ -2950,12 +3024,12 @@ private void validate() {
int maxInstantsToKeep = Integer.parseInt(writeConfig.getStringOrDefault(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP));
int cleanerCommitsRetained =
Integer.parseInt(writeConfig.getStringOrDefault(HoodieCleanConfig.CLEANER_COMMITS_RETAINED));
- ValidationUtils.checkArgument(maxInstantsToKeep > minInstantsToKeep,
+ checkArgument(maxInstantsToKeep > minInstantsToKeep,
String.format(
"Increase %s=%d to be greater than %s=%d.",
HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), maxInstantsToKeep,
HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), minInstantsToKeep));
- ValidationUtils.checkArgument(minInstantsToKeep > cleanerCommitsRetained,
+ checkArgument(minInstantsToKeep > cleanerCommitsRetained,
String.format(
"Increase %s=%d to be greater than %s=%d. Otherwise, there is risk of incremental pull "
+ "missing data from few instants.",
@@ -2964,14 +3038,21 @@ private void validate() {
boolean inlineCompact = writeConfig.getBoolean(HoodieCompactionConfig.INLINE_COMPACT);
boolean inlineCompactSchedule = writeConfig.getBoolean(HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT);
- ValidationUtils.checkArgument(!(inlineCompact && inlineCompactSchedule), String.format("Either of inline compaction (%s) or "
+ checkArgument(!(inlineCompact && inlineCompactSchedule), String.format("Either of inline compaction (%s) or "
+ "schedule inline compaction (%s) can be enabled. Both can't be set to true at the same time. %s, %s", HoodieCompactionConfig.INLINE_COMPACT.key(),
HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT.key(), inlineCompact, inlineCompactSchedule));
}
public HoodieWriteConfig build() {
+ return build(true);
+ }
+
+ @VisibleForTesting
+ public HoodieWriteConfig build(boolean shouldValidate) {
setDefaults();
- validate();
+ if (shouldValidate) {
+ validate();
+ }
// Build WriteConfig at the end
return new HoodieWriteConfig(engineType, writeConfig.getProps());
}
@@ -2982,7 +3063,9 @@ private String getDefaultMarkersType(EngineType engineType) {
if (writeConfig.isEmbeddedTimelineServerEnabled()) {
return MarkerType.TIMELINE_SERVER_BASED.toString();
} else {
- LOG.warn("Embedded timeline server is disabled, fallback to use direct marker type for spark");
+ if (!HoodieTableMetadata.isMetadataTable(writeConfig.getBasePath())) {
+ LOG.warn("Embedded timeline server is disabled, fallback to use direct marker type for spark");
+ }
return MarkerType.DIRECT.toString();
}
case FLINK:
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/HoodieLazyInsertIterable.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/HoodieLazyInsertIterable.java
index 20f75f63c5234..8cc8c07a02ec2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/HoodieLazyInsertIterable.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/HoodieLazyInsertIterable.java
@@ -21,13 +21,13 @@
import org.apache.avro.Schema;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.utils.LazyIterableIterator;
-import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.io.CreateHandleFactory;
import org.apache.hudi.io.WriteHandleFactory;
import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.util.ExecutorFactory;
import java.util.Iterator;
import java.util.List;
@@ -90,23 +90,25 @@ public R getResult() {
}
}
- static Function, HoodieInsertValueGenResult> getCloningTransformer(Schema schema,
- HoodieWriteConfig config) {
- return getCloningTransformerInternal(schema, config.getProps());
+ static Function, HoodieInsertValueGenResult> getTransformer(Schema schema,
+ HoodieWriteConfig writeConfig) {
+ return getTransformerInternal(schema, writeConfig);
}
- static Function, HoodieInsertValueGenResult> getCloningTransformer(Schema schema) {
- return getCloningTransformerInternal(schema, new TypedProperties());
- }
+ private static Function, HoodieInsertValueGenResult> getTransformerInternal(Schema schema,
+ HoodieWriteConfig writeConfig) {
+ // NOTE: Whether record have to be cloned here is determined based on the executor type used
+ // for writing: executors relying on an inner queue, will be keeping references to the records
+ // and therefore in the environments where underlying buffer holding the record could be
+ // reused (for ex, Spark) we need to make sure that we get a clean copy of
+ // it since these records will be subsequently buffered (w/in the in-memory queue);
+ // Only case when we don't need to make a copy is when using [[SimpleExecutor]] which
+ // is guaranteed to not hold on to references to any records
+ boolean shouldClone = ExecutorFactory.isBufferingRecords(writeConfig);
- private static Function, HoodieInsertValueGenResult> getCloningTransformerInternal(Schema schema,
- TypedProperties props) {
return record -> {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be subsequently buffered (w/in the in-memory queue)
- HoodieRecord clonedRecord = record.copy();
- return new HoodieInsertValueGenResult(clonedRecord, schema, props);
+ HoodieRecord clonedRecord = shouldClone ? record.copy() : record;
+ return new HoodieInsertValueGenResult(clonedRecord, schema, writeConfig.getProps());
};
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
index 8a8a03e1b17ac..2ec46f19a43cc 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
@@ -62,9 +62,8 @@ public class HoodieIndexUtils {
* @param hoodieTable Instance of {@link HoodieTable} of interest
* @return the list of {@link HoodieBaseFile}
*/
- public static List getLatestBaseFilesForPartition(
- final String partition,
- final HoodieTable hoodieTable) {
+ public static List getLatestBaseFilesForPartition(String partition,
+ HoodieTable hoodieTable) {
Option latestCommitTime = hoodieTable.getMetaClient().getCommitsTimeline()
.filterCompletedInstants().lastInstant();
if (latestCommitTime.isPresent()) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
index 9430d9bb5e50b..f144540ed2280 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
@@ -19,12 +19,11 @@
package org.apache.hudi.index.bloom;
-import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecordLocation;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieTable;
@@ -51,7 +50,7 @@ public abstract class BaseHoodieBloomIndexHelper implements Serializable {
public abstract HoodiePairData findMatchingFilesForRecordKeys(
HoodieWriteConfig config, HoodieEngineContext context, HoodieTable hoodieTable,
HoodiePairData partitionRecordKeyPairs,
- HoodieData> fileComparisonPairs,
+ HoodiePairData fileComparisonPairs,
Map> partitionToFileInfo,
Map recordsPerPartition);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
index 57d9def9b4201..833db3b1c75a1 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
@@ -27,6 +27,7 @@
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
@@ -35,7 +36,6 @@
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieMetadataException;
import org.apache.hudi.exception.MetadataNotFoundException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.index.HoodieIndexUtils;
@@ -45,9 +45,9 @@
import org.apache.log4j.Logger;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.util.stream.Collectors.groupingBy;
@@ -91,7 +91,8 @@ record -> new ImmutablePair<>(record.getPartitionPath(), record.getRecordKey()))
// Cache the result, for subsequent stages.
if (config.getBloomIndexUseCaching()) {
- keyFilenamePairs.persist("MEMORY_AND_DISK_SER");
+ keyFilenamePairs.persist(new HoodieConfig(config.getProps())
+ .getString(HoodieIndexConfig.BLOOM_INDEX_INPUT_STORAGE_LEVEL_VALUE));
}
if (LOG.isDebugEnabled()) {
long totalTaggedRecords = keyFilenamePairs.count();
@@ -127,7 +128,7 @@ private HoodiePairData lookupIndex(
// Step 3: Obtain a HoodieData, for each incoming record, that already exists, with the file id,
// that contains it.
- HoodieData> fileComparisonPairs =
+ HoodiePairData fileComparisonPairs =
explodeRecordsWithFileComparisons(partitionToFileInfo, partitionRecordKeyPairs);
return bloomIndexHelper.findMatchingFilesForRecordKeys(config, context, hoodieTable,
@@ -210,34 +211,35 @@ protected List> loadColumnRangesFromMetaIndex(
// also obtain file ranges, if range pruning is enabled
context.setJobStatus(this.getClass().getName(), "Load meta index key ranges for file slices: " + config.getTableName());
- final String keyField = hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();
- return context.flatMap(partitions, partitionName -> {
- // Partition and file name pairs
- List> partitionFileNameList = HoodieIndexUtils.getLatestBaseFilesForPartition(partitionName,
- hoodieTable).stream().map(baseFile -> Pair.of(partitionName, baseFile.getFileName()))
- .sorted()
- .collect(toList());
- if (partitionFileNameList.isEmpty()) {
- return Stream.empty();
- }
- try {
- Map, HoodieMetadataColumnStats> fileToColumnStatsMap =
- hoodieTable.getMetadataTable().getColumnStats(partitionFileNameList, keyField);
- List> result = new ArrayList<>();
- for (Map.Entry, HoodieMetadataColumnStats> entry : fileToColumnStatsMap.entrySet()) {
- result.add(Pair.of(entry.getKey().getLeft(),
- new BloomIndexFileInfo(
- FSUtils.getFileId(entry.getKey().getRight()),
- // NOTE: Here we assume that the type of the primary key field is string
- (String) unwrapStatisticValueWrapper(entry.getValue().getMinValue()),
- (String) unwrapStatisticValueWrapper(entry.getValue().getMaxValue())
- )));
- }
- return result.stream();
- } catch (MetadataNotFoundException me) {
- throw new HoodieMetadataException("Unable to find column range metadata for partition:" + partitionName, me);
- }
- }, Math.max(partitions.size(), 1));
+ String keyField = hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();
+
+ // Partition and file name pairs
+ List> partitionFileNameList =
+ HoodieIndexUtils.getLatestBaseFilesForAllPartitions(partitions, context, hoodieTable).stream()
+ .map(partitionBaseFilePair -> Pair.of(partitionBaseFilePair.getLeft(), partitionBaseFilePair.getRight().getFileName()))
+ .sorted()
+ .collect(toList());
+
+ if (partitionFileNameList.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ Map, HoodieMetadataColumnStats> fileToColumnStatsMap =
+ hoodieTable.getMetadataTable().getColumnStats(partitionFileNameList, keyField);
+
+ List> result = new ArrayList<>(fileToColumnStatsMap.size());
+
+ for (Map.Entry, HoodieMetadataColumnStats> entry : fileToColumnStatsMap.entrySet()) {
+ result.add(Pair.of(entry.getKey().getLeft(),
+ new BloomIndexFileInfo(
+ FSUtils.getFileId(entry.getKey().getRight()),
+ // NOTE: Here we assume that the type of the primary key field is string
+ (String) unwrapStatisticValueWrapper(entry.getValue().getMinValue()),
+ (String) unwrapStatisticValueWrapper(entry.getValue().getMaxValue())
+ )));
+ }
+
+ return result;
}
@Override
@@ -278,7 +280,7 @@ public boolean isImplicitWithStorage() {
* Sub-partition to ensure the records can be looked up against files & also prune file<=>record comparisons based on
* recordKey ranges in the index info.
*/
- HoodieData> explodeRecordsWithFileComparisons(
+ HoodiePairData explodeRecordsWithFileComparisons(
final Map> partitionToFileIndexInfo,
HoodiePairData partitionRecordKeyPairs) {
IndexFileFilter indexFileFilter =
@@ -289,11 +291,13 @@ HoodieData> explodeRecordsWithFileComparisons(
String recordKey = partitionRecordKeyPair.getRight();
String partitionPath = partitionRecordKeyPair.getLeft();
- return indexFileFilter.getMatchingFilesAndPartition(partitionPath, recordKey).stream()
- .map(partitionFileIdPair -> (Pair) new ImmutablePair<>(partitionFileIdPair.getRight(),
- new HoodieKey(recordKey, partitionPath)))
- .collect(Collectors.toList());
- }).flatMap(List::iterator);
+ return indexFileFilter.getMatchingFilesAndPartition(partitionPath, recordKey)
+ .stream()
+ .map(partitionFileIdPair ->
+ new ImmutablePair<>(
+ new HoodieFileGroupId(partitionFileIdPair.getLeft(), partitionFileIdPair.getRight()), recordKey));
+ })
+ .flatMapToPair(Stream::iterator);
}
/**
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
similarity index 61%
rename from hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
index 80031f4e8f025..52b504e9ab168 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
@@ -19,7 +19,8 @@
package org.apache.hudi.index.bloom;
import org.apache.hudi.client.utils.LazyIterableIterator;
-import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.function.SerializableFunction;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
@@ -28,53 +29,67 @@
import org.apache.hudi.io.HoodieKeyLookupResult;
import org.apache.hudi.table.HoodieTable;
-import java.util.function.Function;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.function.Function;
/**
- * Function performing actual checking of list containing (fileId, hoodieKeys) against the actual files.
+ * Function accepting a tuple of {@link HoodieFileGroupId} and a record key and producing
+ * a list of {@link HoodieKeyLookupResult} for every file identified by the file-group ids
+ *
+ * @param type of the tuple of {@code (HoodieFileGroupId, )}. Note that this is
+ * parameterized as generic such that this code could be reused for Spark as well
*/
-public class HoodieBaseBloomIndexCheckFunction
- implements Function>, Iterator>> {
+public class HoodieBloomIndexCheckFunction
+ implements Function, Iterator>>, Serializable {
private final HoodieTable hoodieTable;
private final HoodieWriteConfig config;
- public HoodieBaseBloomIndexCheckFunction(HoodieTable hoodieTable, HoodieWriteConfig config) {
+ private final SerializableFunction fileGroupIdExtractor;
+ private final SerializableFunction recordKeyExtractor;
+
+ public HoodieBloomIndexCheckFunction(HoodieTable hoodieTable,
+ HoodieWriteConfig config,
+ SerializableFunction fileGroupIdExtractor,
+ SerializableFunction recordKeyExtractor) {
this.hoodieTable = hoodieTable;
this.config = config;
+ this.fileGroupIdExtractor = fileGroupIdExtractor;
+ this.recordKeyExtractor = recordKeyExtractor;
}
@Override
- public Iterator> apply(Iterator> filePartitionRecordKeyTripletItr) {
- return new LazyKeyCheckIterator(filePartitionRecordKeyTripletItr);
+ public Iterator> apply(Iterator fileGroupIdRecordKeyPairIterator) {
+ return new LazyKeyCheckIterator(fileGroupIdRecordKeyPairIterator);
}
- class LazyKeyCheckIterator extends LazyIterableIterator, List> {
+ protected class LazyKeyCheckIterator extends LazyIterableIterator> {
private HoodieKeyLookupHandle keyLookupHandle;
- LazyKeyCheckIterator(Iterator> filePartitionRecordKeyTripletItr) {
+ LazyKeyCheckIterator(Iterator filePartitionRecordKeyTripletItr) {
super(filePartitionRecordKeyTripletItr);
}
- @Override
- protected void start() {
- }
-
@Override
protected List computeNext() {
+
List ret = new ArrayList<>();
try {
// process one file in each go.
while (inputItr.hasNext()) {
- Pair currentTuple = inputItr.next();
- String fileId = currentTuple.getLeft();
- String partitionPath = currentTuple.getRight().getPartitionPath();
- String recordKey = currentTuple.getRight().getRecordKey();
+ I tuple = inputItr.next();
+
+ HoodieFileGroupId fileGroupId = fileGroupIdExtractor.apply(tuple);
+ String recordKey = recordKeyExtractor.apply(tuple);
+
+ String fileId = fileGroupId.getFileId();
+ String partitionPath = fileGroupId.getPartitionPath();
+
Pair partitionPathFilePair = Pair.of(partitionPath, fileId);
// lazily init state
@@ -100,15 +115,13 @@ protected List computeNext() {
}
} catch (Throwable e) {
if (e instanceof HoodieException) {
- throw e;
+ throw (HoodieException) e;
}
+
throw new HoodieIndexException("Error checking bloom filter index. ", e);
}
- return ret;
- }
- @Override
- protected void end() {
+ return ret;
}
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
index 5f2007ea53668..b5604312d3f7b 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
@@ -25,6 +25,7 @@
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
@@ -41,7 +42,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* This filter will only work with hoodie table since it will only load partitions
@@ -74,7 +75,7 @@ List> loadColumnRangesFromFiles(List pa
*/
@Override
- HoodieData> explodeRecordsWithFileComparisons(
+ HoodiePairData explodeRecordsWithFileComparisons(
final Map> partitionToFileIndexInfo,
HoodiePairData partitionRecordKeyPairs) {
@@ -87,10 +88,11 @@ HoodieData> explodeRecordsWithFileComparisons(
String partitionPath = partitionRecordKeyPair.getLeft();
return indexFileFilter.getMatchingFilesAndPartition(partitionPath, recordKey).stream()
- .map(partitionFileIdPair -> (Pair) new ImmutablePair<>(partitionFileIdPair.getRight(),
- new HoodieKey(recordKey, partitionFileIdPair.getLeft())))
- .collect(Collectors.toList());
- }).flatMap(List::iterator);
+ .map(partitionFileIdPair ->
+ new ImmutablePair<>(
+ new HoodieFileGroupId(partitionFileIdPair.getLeft(), partitionFileIdPair.getRight()), recordKey));
+ })
+ .flatMapToPair(Stream::iterator);
}
/**
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
index cffee5ee74081..b47f5cf066c54 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
@@ -19,20 +19,20 @@
package org.apache.hudi.index.bloom;
-import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.io.HoodieKeyLookupResult;
import org.apache.hudi.table.HoodieTable;
-import java.util.ArrayList;
+import java.util.Collection;
import java.util.Comparator;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -56,21 +56,21 @@ public static ListBasedHoodieBloomIndexHelper getInstance() {
public HoodiePairData findMatchingFilesForRecordKeys(
HoodieWriteConfig config, HoodieEngineContext context, HoodieTable hoodieTable,
HoodiePairData partitionRecordKeyPairs,
- HoodieData> fileComparisonPairs,
+ HoodiePairData fileComparisonPairs,
Map> partitionToFileInfo, Map recordsPerPartition) {
- List> fileComparisonPairList =
+ List> fileComparisonPairList =
fileComparisonPairs.collectAsList().stream()
.sorted(Comparator.comparing(Pair::getLeft)).collect(toList());
- List keyLookupResults = new ArrayList<>();
- Iterator> iterator = new HoodieBaseBloomIndexCheckFunction(
- hoodieTable, config).apply(fileComparisonPairList.iterator());
- while (iterator.hasNext()) {
- keyLookupResults.addAll(iterator.next());
- }
+ List keyLookupResults =
+ CollectionUtils.toStream(
+ new HoodieBloomIndexCheckFunction>(hoodieTable, config, Pair::getLeft, Pair::getRight)
+ .apply(fileComparisonPairList.iterator())
+ )
+ .flatMap(Collection::stream)
+ .filter(lr -> lr.getMatchingRecordKeys().size() > 0)
+ .collect(toList());
- keyLookupResults = keyLookupResults.stream().filter(
- lr -> lr.getMatchingRecordKeys().size() > 0).collect(toList());
return context.parallelize(keyLookupResults).flatMap(lookupResult ->
lookupResult.getMatchingRecordKeys().stream()
.map(recordKey -> new ImmutablePair<>(lookupResult, recordKey)).iterator()
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
index 35f9205a8e58b..5264c8b39f192 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
@@ -76,7 +76,11 @@ public static String partitionBucketIdStr(String partition, int bucketId) {
}
public static int bucketIdFromFileId(String fileId) {
- return Integer.parseInt(fileId.substring(0, 8));
+ return Integer.parseInt(bucketIdStrFromFileId(fileId));
+ }
+
+ public static String bucketIdStrFromFileId(String fileId) {
+ return fileId.substring(0, 8);
}
public static String bucketIdStr(int n) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
index 02053127d9c83..a69d460334362 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
@@ -114,9 +114,9 @@ public class HoodieAppendHandle extends HoodieWriteHandle header = new HashMap<>();
private SizeEstimator sizeEstimator;
@@ -236,21 +236,25 @@ private Option prepareRecord(HoodieRecord hoodieRecord) {
// If the format can not record the operation field, nullify the DELETE payload manually.
boolean nullifyPayload = HoodieOperation.isDelete(hoodieRecord.getOperation()) && !config.allowOperationMetadataField();
recordProperties.put(HoodiePayloadProps.PAYLOAD_IS_UPDATE_RECORD_FOR_MOR, String.valueOf(isUpdateRecord));
- Option finalRecord = nullifyPayload ? Option.empty() : Option.of(hoodieRecord);
+
+ Option finalRecordOpt = nullifyPayload ? Option.empty() : Option.of(hoodieRecord);
// Check for delete
- if (finalRecord.isPresent() && !finalRecord.get().isDelete(schema, recordProperties)) {
- // Check for ignore ExpressionPayload
- if (finalRecord.get().shouldIgnore(schema, recordProperties)) {
- return finalRecord;
+ if (finalRecordOpt.isPresent() && !finalRecordOpt.get().isDelete(schema, recordProperties)) {
+ HoodieRecord finalRecord = finalRecordOpt.get();
+ // Check if the record should be ignored (special case for [[ExpressionPayload]])
+ if (finalRecord.shouldIgnore(schema, recordProperties)) {
+ return finalRecordOpt;
}
- // Convert GenericRecord to GenericRecord with hoodie commit metadata in schema
- HoodieRecord rewrittenRecord = schemaOnReadEnabled ? finalRecord.get().rewriteRecordWithNewSchema(schema, recordProperties, writeSchemaWithMetaFields)
- : finalRecord.get().rewriteRecord(schema, recordProperties, writeSchemaWithMetaFields);
+
+ // Prepend meta-fields into the record
+ MetadataValues metadataValues = populateMetadataFields(finalRecord);
+ HoodieRecord populatedRecord =
+ finalRecord.prependMetaFields(schema, writeSchemaWithMetaFields, metadataValues, recordProperties);
+
// NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
// payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
// it since these records will be put into the recordList(List).
- HoodieRecord populatedRecord = populateMetadataFields(rewrittenRecord.copy(), writeSchemaWithMetaFields, recordProperties);
- finalRecord = Option.of(populatedRecord);
+ finalRecordOpt = Option.of(populatedRecord.copy());
if (isUpdateRecord || isLogCompaction) {
updatedRecordsWritten++;
} else {
@@ -258,7 +262,7 @@ private Option prepareRecord(HoodieRecord hoodieRecord) {
}
recordsWritten++;
} else {
- finalRecord = Option.empty();
+ finalRecordOpt = Option.empty();
recordsDeleted++;
}
@@ -267,7 +271,7 @@ private Option prepareRecord(HoodieRecord hoodieRecord) {
// part of marking
// record successful.
hoodieRecord.deflate();
- return finalRecord;
+ return finalRecordOpt;
} catch (Exception e) {
LOG.error("Error writing record " + hoodieRecord, e);
writeStatus.markFailure(hoodieRecord, e, recordMetadata);
@@ -275,7 +279,7 @@ private Option prepareRecord(HoodieRecord hoodieRecord) {
return Option.empty();
}
- private HoodieRecord populateMetadataFields(HoodieRecord hoodieRecord, Schema schema, Properties prop) throws IOException {
+ private MetadataValues populateMetadataFields(HoodieRecord hoodieRecord) {
MetadataValues metadataValues = new MetadataValues();
if (config.populateMetaFields()) {
String seqId =
@@ -292,7 +296,7 @@ private HoodieRecord populateMetadataFields(HoodieRecord hoodieRecord, Schema
metadataValues.setOperation(hoodieRecord.getOperation().getName());
}
- return hoodieRecord.updateMetadataValues(schema, prop, metadataValues);
+ return metadataValues;
}
private void initNewStatus() {
@@ -588,7 +592,7 @@ private void flushToDiskIfRequired(HoodieRecord record, boolean appendDeleteBloc
}
// Append if max number of records reached to achieve block size
- if (numberOfRecords >= (int) (maxBlockSize / averageRecordSize)) {
+ if (numberOfRecords >= (long) (maxBlockSize / averageRecordSize)) {
// Recompute averageRecordSize before writing a new block and update existing value with
// avg of new and old
LOG.info("Flush log block to disk, the current avgRecordSize => " + averageRecordSize);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBootstrapHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBootstrapHandle.java
index f110bf585dbe3..e4985907e2e38 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBootstrapHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBootstrapHandle.java
@@ -18,13 +18,19 @@
package org.apache.hudi.io;
-import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.avro.JsonProperties;
+import org.apache.avro.Schema;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieTable;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.avro.AvroSchemaUtils.createNullableSchema;
+
/**
* This class is essentially same as Create Handle but overrides two things
* 1) Schema : Metadata bootstrap writes only metadata fields as part of write. So, setup the writer schema accordingly.
@@ -34,14 +40,28 @@
*/
public class HoodieBootstrapHandle extends HoodieCreateHandle {
+ // NOTE: We have to use schema containing all the meta-fields in here b/c unlike for [[HoodieAvroRecord]],
+ // [[HoodieSparkRecord]] requires records to always bear either all or no meta-fields in the
+ // record schema (ie partial inclusion of the meta-fields in the schema is not allowed)
+ public static final Schema METADATA_BOOTSTRAP_RECORD_SCHEMA = createMetadataBootstrapRecordSchema();
+
public HoodieBootstrapHandle(HoodieWriteConfig config, String commitTime, HoodieTable hoodieTable,
String partitionPath, String fileId, TaskContextSupplier taskContextSupplier) {
super(config, commitTime, hoodieTable, partitionPath, fileId,
- Option.of(HoodieAvroUtils.RECORD_KEY_SCHEMA), taskContextSupplier);
+ Option.of(METADATA_BOOTSTRAP_RECORD_SCHEMA), taskContextSupplier);
}
@Override
public boolean canWrite(HoodieRecord record) {
return true;
}
+
+ private static Schema createMetadataBootstrapRecordSchema() {
+ List fields =
+ HoodieRecord.HOODIE_META_COLUMNS.stream()
+ .map(metaField ->
+ new Schema.Field(metaField, createNullableSchema(Schema.Type.STRING), "", JsonProperties.NULL_VALUE))
+ .collect(Collectors.toList());
+ return Schema.createRecord("HoodieRecordKey", "", "", false, fields);
+ }
}
\ No newline at end of file
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java
index be646df85be4d..d80f4250c60bd 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java
@@ -58,6 +58,9 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode.data_before;
+import static org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode.data_before_after;
+
/**
* This class encapsulates all the cdc-writing functions.
*/
@@ -89,7 +92,7 @@ public class HoodieCDCLogger implements Closeable {
private final CDCTransformer transformer;
// Max block size to limit to for a log block
- private final int maxBlockSize;
+ private final long maxBlockSize;
// Average cdc record size. This size is updated at the end of every log block flushed to disk
private long averageCDCRecordSize = 0;
@@ -240,10 +243,10 @@ public void close() {
// -------------------------------------------------------------------------
private CDCTransformer getTransformer() {
- if (cdcSupplementalLoggingMode.equals(HoodieCDCSupplementalLoggingMode.WITH_BEFORE_AFTER)) {
+ if (cdcSupplementalLoggingMode == data_before_after) {
return (operation, recordKey, oldRecord, newRecord) ->
- HoodieCDCUtils.cdcRecord(cdcSchema, operation.getValue(), commitTime, removeCommitMetadata(oldRecord), newRecord);
- } else if (cdcSupplementalLoggingMode.equals(HoodieCDCSupplementalLoggingMode.WITH_BEFORE)) {
+ HoodieCDCUtils.cdcRecord(cdcSchema, operation.getValue(), commitTime, removeCommitMetadata(oldRecord), removeCommitMetadata(newRecord));
+ } else if (cdcSupplementalLoggingMode == data_before) {
return (operation, recordKey, oldRecord, newRecord) ->
HoodieCDCUtils.cdcRecord(cdcSchema, operation.getValue(), recordKey, removeCommitMetadata(oldRecord));
} else {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
index 96611959f9a5a..4d3f52a5ba6aa 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
@@ -136,24 +136,22 @@ protected void doWrite(HoodieRecord record, Schema schema, TypedProperties props
if (record.shouldIgnore(schema, config.getProps())) {
return;
}
- // Convert GenericRecord to GenericRecord with hoodie commit metadata in schema
- HoodieRecord rewriteRecord;
- if (schemaOnReadEnabled) {
- rewriteRecord = record.rewriteRecordWithNewSchema(schema, config.getProps(), writeSchemaWithMetaFields);
- } else {
- rewriteRecord = record.rewriteRecord(schema, config.getProps(), writeSchemaWithMetaFields);
- }
+
MetadataValues metadataValues = new MetadataValues().setFileName(path.getName());
- rewriteRecord = rewriteRecord.updateMetadataValues(writeSchemaWithMetaFields, config.getProps(), metadataValues);
+ HoodieRecord populatedRecord =
+ record.prependMetaFields(schema, writeSchemaWithMetaFields, metadataValues, config.getProps());
+
if (preserveMetadata) {
- fileWriter.write(record.getRecordKey(), rewriteRecord, writeSchemaWithMetaFields);
+ fileWriter.write(record.getRecordKey(), populatedRecord, writeSchemaWithMetaFields);
} else {
- fileWriter.writeWithMetadata(record.getKey(), rewriteRecord, writeSchemaWithMetaFields);
+ fileWriter.writeWithMetadata(record.getKey(), populatedRecord, writeSchemaWithMetaFields);
}
- // update the new location of record, so we know where to find it next
+
+ // Update the new location of record, so we know where to find it next
record.unseal();
record.setNewLocation(new HoodieRecordLocation(instantTime, writeStatus.getFileId()));
record.seal();
+
recordsWritten++;
insertRecordsWritten++;
} else {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
index 0460f88101c54..f92a1e73d5849 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
@@ -374,20 +374,16 @@ public void write(HoodieRecord oldRecord) {
}
protected void writeToFile(HoodieKey key, HoodieRecord record, Schema schema, Properties prop, boolean shouldPreserveRecordMetadata) throws IOException {
- HoodieRecord rewriteRecord;
- if (schemaOnReadEnabled) {
- rewriteRecord = record.rewriteRecordWithNewSchema(schema, prop, writeSchemaWithMetaFields);
- } else {
- rewriteRecord = record.rewriteRecord(schema, prop, writeSchemaWithMetaFields);
- }
// NOTE: `FILENAME_METADATA_FIELD` has to be rewritten to correctly point to the
// file holding this record even in cases when overall metadata is preserved
MetadataValues metadataValues = new MetadataValues().setFileName(newFilePath.getName());
- rewriteRecord = rewriteRecord.updateMetadataValues(writeSchemaWithMetaFields, prop, metadataValues);
+ HoodieRecord populatedRecord =
+ record.prependMetaFields(schema, writeSchemaWithMetaFields, metadataValues, prop);
+
if (shouldPreserveRecordMetadata) {
- fileWriter.write(key.getRecordKey(), rewriteRecord, writeSchemaWithMetaFields);
+ fileWriter.write(key.getRecordKey(), populatedRecord, writeSchemaWithMetaFields);
} else {
- fileWriter.writeWithMetadata(key, rewriteRecord, writeSchemaWithMetaFields);
+ fileWriter.writeWithMetadata(key, populatedRecord, writeSchemaWithMetaFields);
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
index b110c2c081782..8aadd637f0e52 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
@@ -80,21 +80,11 @@ public static HoodieMergeHandle create(
TaskContextSupplier taskContextSupplier,
Option keyGeneratorOpt) {
if (table.requireSortedRecords()) {
- if (table.getMetaClient().getTableConfig().isCDCEnabled()) {
- return new HoodieSortedMergeHandleWithChangeLog<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- } else {
- return new HoodieSortedMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- }
+ return new HoodieSortedMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
+ dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
} else {
- if (table.getMetaClient().getTableConfig().isCDCEnabled()) {
- return new HoodieMergeHandleWithChangeLog<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- } else {
- return new HoodieMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- }
+ return new HoodieMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
+ dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
}
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java
index 8e470471db91a..889d7a64769a4 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java
@@ -95,7 +95,7 @@ protected HoodieWriteHandle(HoodieWriteConfig config, String instantTime, String
!hoodieTable.getIndex().isImplicitWithStorage(), config.getWriteStatusFailureFraction());
this.taskContextSupplier = taskContextSupplier;
this.writeToken = makeWriteToken();
- schemaOnReadEnabled = !isNullOrEmpty(hoodieTable.getConfig().getInternalSchema());
+ this.schemaOnReadEnabled = !isNullOrEmpty(hoodieTable.getConfig().getInternalSchema());
this.recordMerger = config.getRecordMerger();
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
index bbfa4460af411..78c4b1fc44192 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
@@ -26,6 +26,7 @@
import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.client.BaseHoodieWriteClient;
+import org.apache.hudi.client.FailOnFirstErrorWriteStatus;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.data.HoodieData;
@@ -70,7 +71,7 @@
import org.apache.hudi.exception.HoodieMetadataException;
import org.apache.hudi.hadoop.CachingPath;
import org.apache.hudi.hadoop.SerializablePath;
-
+import org.apache.hudi.table.action.compact.strategy.UnBoundedCompactionStrategy;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
@@ -93,6 +94,7 @@
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy.EAGER;
import static org.apache.hudi.common.table.HoodieTableConfig.ARCHIVELOG_FOLDER;
import static org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.getIndexInflightInstant;
@@ -100,6 +102,7 @@
import static org.apache.hudi.common.util.StringUtils.EMPTY_STRING;
import static org.apache.hudi.metadata.HoodieTableMetadata.METADATA_TABLE_NAME_SUFFIX;
import static org.apache.hudi.metadata.HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.METADATA_INDEXER_TIME_SUFFIX;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getInflightAndCompletedMetadataPartitions;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getInflightMetadataPartitions;
@@ -134,15 +137,17 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
/**
* Hudi backed table metadata writer.
*
- * @param hadoopConf - Hadoop configuration to use for the metadata writer
- * @param writeConfig - Writer config
- * @param engineContext - Engine context
- * @param actionMetadata - Optional action metadata to help decide initialize operations
- * @param - Action metadata types extending Avro generated SpecificRecordBase
- * @param inflightInstantTimestamp - Timestamp of any instant in progress
+ * @param hadoopConf Hadoop configuration to use for the metadata writer
+ * @param writeConfig Writer config
+ * @param failedWritesCleaningPolicy Cleaning policy on failed writes
+ * @param engineContext Engine context
+ * @param actionMetadata Optional action metadata to help decide initialize operations
+ * @param Action metadata types extending Avro generated SpecificRecordBase
+ * @param inflightInstantTimestamp Timestamp of any instant in progress
*/
protected HoodieBackedTableMetadataWriter(Configuration hadoopConf,
HoodieWriteConfig writeConfig,
+ HoodieFailedWritesCleaningPolicy failedWritesCleaningPolicy,
HoodieEngineContext engineContext,
Option actionMetadata,
Option inflightInstantTimestamp) {
@@ -154,7 +159,7 @@ protected HoodieBackedTableMetadataWriter(Configu
if (writeConfig.isMetadataTableEnabled()) {
this.tableName = writeConfig.getTableName() + METADATA_TABLE_NAME_SUFFIX;
- this.metadataWriteConfig = createMetadataWriteConfig(writeConfig);
+ this.metadataWriteConfig = createMetadataWriteConfig(writeConfig, failedWritesCleaningPolicy);
enabled = true;
// Inline compaction and auto clean is required as we dont expose this table outside
@@ -162,9 +167,12 @@ protected HoodieBackedTableMetadataWriter(Configu
"Cleaning is controlled internally for Metadata table.");
ValidationUtils.checkArgument(!this.metadataWriteConfig.inlineCompactionEnabled(),
"Compaction is controlled internally for metadata table.");
- // Metadata Table cannot have metadata listing turned on. (infinite loop, much?)
+ // Auto commit is required
ValidationUtils.checkArgument(this.metadataWriteConfig.shouldAutoCommit(),
"Auto commit is required for Metadata Table");
+ ValidationUtils.checkArgument(this.metadataWriteConfig.getWriteStatusClassName().equals(FailOnFirstErrorWriteStatus.class.getName()),
+ "MDT should use " + FailOnFirstErrorWriteStatus.class.getName());
+ // Metadata Table cannot have metadata listing turned on. (infinite loop, much?)
ValidationUtils.checkArgument(!this.metadataWriteConfig.isMetadataTableEnabled(),
"File listing cannot be used for Metadata Table");
@@ -181,7 +189,7 @@ protected HoodieBackedTableMetadataWriter(Configu
public HoodieBackedTableMetadataWriter(Configuration hadoopConf, HoodieWriteConfig writeConfig,
HoodieEngineContext engineContext) {
- this(hadoopConf, writeConfig, engineContext, Option.empty(), Option.empty());
+ this(hadoopConf, writeConfig, EAGER, engineContext, Option.empty(), Option.empty());
}
/**
@@ -232,11 +240,14 @@ private void enablePartition(final MetadataPartitionType partitionType, final Ho
protected abstract void initRegistry();
/**
- * Create a {@code HoodieWriteConfig} to use for the Metadata Table.
+ * Create a {@code HoodieWriteConfig} to use for the Metadata Table. This is used by async
+ * indexer only.
*
- * @param writeConfig {@code HoodieWriteConfig} of the main dataset writer
+ * @param writeConfig {@code HoodieWriteConfig} of the main dataset writer
+ * @param failedWritesCleaningPolicy Cleaning policy on failed writes
*/
- private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfig) {
+ private HoodieWriteConfig createMetadataWriteConfig(
+ HoodieWriteConfig writeConfig, HoodieFailedWritesCleaningPolicy failedWritesCleaningPolicy) {
int parallelism = writeConfig.getMetadataInsertParallelism();
int minCommitsToKeep = Math.max(writeConfig.getMetadataMinCommitsToKeep(), writeConfig.getMinCommitsToKeep());
@@ -268,7 +279,7 @@ private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfi
.withAutoClean(false)
.withCleanerParallelism(parallelism)
.withCleanerPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS)
- .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.EAGER)
+ .withFailedWritesCleaningPolicy(failedWritesCleaningPolicy)
.retainCommits(HoodieMetadataConfig.CLEANER_COMMITS_RETAINED.defaultValue())
.build())
// we will trigger archive manually, to ensure only regular writer invokes it
@@ -283,7 +294,11 @@ private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfi
// by default, the HFile does not keep the metadata fields, set up as false
// to always use the metadata of the new record.
.withPreserveCommitMetadata(false)
- .withLogRecordReaderScanV2(String.valueOf(writeConfig.useScanV2ForLogRecordReader()))
+ .withEnableOptimizedLogBlocksScan(String.valueOf(writeConfig.enableOptimizedLogBlocksScan()))
+ // Compaction on metadata table is used as a barrier for archiving on main dataset and for validating the
+ // deltacommits having corresponding completed commits. Therefore, we need to compact all fileslices of all
+ // partitions together requiring UnBoundedCompactionStrategy.
+ .withCompactionStrategy(new UnBoundedCompactionStrategy())
.build())
.withParallelism(parallelism, parallelism)
.withDeleteParallelism(parallelism)
@@ -291,7 +306,10 @@ private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfi
.withFinalizeWriteParallelism(parallelism)
.withAllowMultiWriteOnSameInstant(true)
.withKeyGenerator(HoodieTableMetadataKeyGenerator.class.getCanonicalName())
- .withPopulateMetaFields(HoodieMetadataConfig.POPULATE_META_FIELDS.defaultValue());
+ .withPopulateMetaFields(HoodieMetadataConfig.POPULATE_META_FIELDS.defaultValue())
+ .withWriteStatusClass(FailOnFirstErrorWriteStatus.class)
+ .withReleaseResourceEnabled(writeConfig.areReleaseResourceEnabled());
+
// RecordKey properties are needed for the metadata table records
final Properties properties = new Properties();
@@ -648,12 +666,9 @@ private List listAllPartitions(HoodieTableMetaClient datasetMetaC
for (DirectoryInfo dirInfo : processedDirectories) {
if (!dirFilterRegex.isEmpty()) {
final String relativePath = dirInfo.getRelativePath();
- if (!relativePath.isEmpty()) {
- Path partitionPath = new Path(datasetBasePath, relativePath);
- if (partitionPath.getName().matches(dirFilterRegex)) {
- LOG.info("Ignoring directory " + partitionPath + " which matches the filter regex " + dirFilterRegex);
- continue;
- }
+ if (!relativePath.isEmpty() && relativePath.matches(dirFilterRegex)) {
+ LOG.info("Ignoring directory " + relativePath + " which matches the filter regex " + dirFilterRegex);
+ continue;
}
}
@@ -875,7 +890,7 @@ public void buildMetadataPartitions(HoodieEngineContext engineContext, List new HoodieMetadataException("No completed deltacommit in metadata table"))
.getTimestamp();
+ // we need to find if there are any inflights in data table timeline before or equal to the latest delta commit in metadata table.
+ // Whenever you want to change this logic, please ensure all below scenarios are considered.
+ // a. There could be a chance that latest delta commit in MDT is committed in MDT, but failed in DT. And so findInstantsBeforeOrEquals() should be employed
+ // b. There could be DT inflights after latest delta commit in MDT and we are ok with it. bcoz, the contract is, latest compaction instant time in MDT represents
+ // any instants before that is already synced with metadata table.
+ // c. Do consider out of order commits. For eg, c4 from DT could complete before c3. and we can't trigger compaction in MDT with c4 as base instant time, until every
+ // instant before c4 is synced with metadata table.
List pendingInstants = dataMetaClient.reloadActiveTimeline().filterInflightsAndRequested()
- .findInstantsBefore(latestDeltaCommitTimeInMetadataTable).getInstants();
+ .findInstantsBeforeOrEquals(latestDeltaCommitTimeInMetadataTable).getInstants();
if (!pendingInstants.isEmpty()) {
LOG.info(String.format(
@@ -1069,6 +1091,7 @@ protected void cleanIfNecessary(BaseHoodieWriteClient writeClient, String instan
// delta commits synced over will not have an instant time lesser than the last completed instant on the
// metadata table.
writeClient.clean(instantTime + "002");
+ writeClient.lazyRollbackFailedIndexing();
}
/**
@@ -1086,39 +1109,45 @@ private void initialCommit(String createInstantTime, List
Map> partitionToRecordsMap = new HashMap<>();
- List partitionInfoList = listAllPartitions(dataMetaClient);
- Map> partitionToFilesMap = partitionInfoList.stream()
- .map(p -> {
- String partitionName = HoodieTableMetadataUtil.getPartitionIdentifier(p.getRelativePath());
- return Pair.of(partitionName, p.getFileNameToSizeMap());
- })
- .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
-
- int totalDataFilesCount = partitionToFilesMap.values().stream().mapToInt(Map::size).sum();
- List partitions = new ArrayList<>(partitionToFilesMap.keySet());
-
- if (partitionTypes.contains(MetadataPartitionType.FILES)) {
- // Record which saves the list of all partitions
- HoodieRecord allPartitionRecord = HoodieMetadataPayload.createPartitionListRecord(partitions);
- HoodieData filesPartitionRecords = getFilesPartitionRecords(createInstantTime, partitionInfoList, allPartitionRecord);
- ValidationUtils.checkState(filesPartitionRecords.count() == (partitions.size() + 1));
- partitionToRecordsMap.put(MetadataPartitionType.FILES, filesPartitionRecords);
- }
+ // skip file system listing to populate metadata records if its a fresh table.
+ // this is applicable only if the table already has N commits and metadata is enabled at a later point in time.
+ if (createInstantTime.equals(SOLO_COMMIT_TIMESTAMP)) { // SOLO_COMMIT_TIMESTAMP will be the initial commit time in MDT for a fresh table.
+ // If not, last completed commit in data table will be chosen as the initial commit time.
+ LOG.info("Triggering empty Commit to metadata to initialize");
+ } else {
+ List partitionInfoList = listAllPartitions(dataMetaClient);
+ Map> partitionToFilesMap = partitionInfoList.stream()
+ .map(p -> {
+ String partitionName = HoodieTableMetadataUtil.getPartitionIdentifier(p.getRelativePath());
+ return Pair.of(partitionName, p.getFileNameToSizeMap());
+ })
+ .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
+
+ int totalDataFilesCount = partitionToFilesMap.values().stream().mapToInt(Map::size).sum();
+ List partitions = new ArrayList<>(partitionToFilesMap.keySet());
+
+ if (partitionTypes.contains(MetadataPartitionType.FILES)) {
+ // Record which saves the list of all partitions
+ HoodieRecord allPartitionRecord = HoodieMetadataPayload.createPartitionListRecord(partitions);
+ HoodieData filesPartitionRecords = getFilesPartitionRecords(createInstantTime, partitionInfoList, allPartitionRecord);
+ ValidationUtils.checkState(filesPartitionRecords.count() == (partitions.size() + 1));
+ partitionToRecordsMap.put(MetadataPartitionType.FILES, filesPartitionRecords);
+ }
- if (partitionTypes.contains(MetadataPartitionType.BLOOM_FILTERS) && totalDataFilesCount > 0) {
- final HoodieData recordsRDD = HoodieTableMetadataUtil.convertFilesToBloomFilterRecords(
- engineContext, Collections.emptyMap(), partitionToFilesMap, getRecordsGenerationParams(), createInstantTime);
- partitionToRecordsMap.put(MetadataPartitionType.BLOOM_FILTERS, recordsRDD);
- }
+ if (partitionTypes.contains(MetadataPartitionType.BLOOM_FILTERS) && totalDataFilesCount > 0) {
+ final HoodieData recordsRDD = HoodieTableMetadataUtil.convertFilesToBloomFilterRecords(
+ engineContext, Collections.emptyMap(), partitionToFilesMap, getRecordsGenerationParams(), createInstantTime);
+ partitionToRecordsMap.put(MetadataPartitionType.BLOOM_FILTERS, recordsRDD);
+ }
- if (partitionTypes.contains(MetadataPartitionType.COLUMN_STATS) && totalDataFilesCount > 0) {
- final HoodieData recordsRDD = HoodieTableMetadataUtil.convertFilesToColumnStatsRecords(
- engineContext, Collections.emptyMap(), partitionToFilesMap, getRecordsGenerationParams());
- partitionToRecordsMap.put(MetadataPartitionType.COLUMN_STATS, recordsRDD);
+ if (partitionTypes.contains(MetadataPartitionType.COLUMN_STATS) && totalDataFilesCount > 0) {
+ final HoodieData recordsRDD = HoodieTableMetadataUtil.convertFilesToColumnStatsRecords(
+ engineContext, Collections.emptyMap(), partitionToFilesMap, getRecordsGenerationParams());
+ partitionToRecordsMap.put(MetadataPartitionType.COLUMN_STATS, recordsRDD);
+ }
+ LOG.info("Committing " + partitions.size() + " partitions and " + totalDataFilesCount + " files to metadata");
}
- LOG.info("Committing " + partitions.size() + " partitions and " + totalDataFilesCount + " files to metadata");
-
commit(createInstantTime, partitionToRecordsMap, false);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
index af8d8d2326187..9800cf268ac4e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
@@ -18,6 +18,7 @@
package org.apache.hudi.table;
+import org.apache.hudi.avro.AvroSchemaUtils;
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.avro.model.HoodieCleanMetadata;
import org.apache.hudi.avro.model.HoodieCleanerPlan;
@@ -41,6 +42,7 @@
import org.apache.hudi.common.fs.ConsistencyGuardConfig;
import org.apache.hudi.common.fs.FailSafeConsistencyGuard;
import org.apache.hudi.common.fs.OptimisticConsistencyGuard;
+import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieTableType;
@@ -91,6 +93,10 @@
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -99,7 +105,8 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.apache.hudi.avro.AvroSchemaUtils.isSchemaCompatible;
+import static org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy.EAGER;
+import static org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy.LAZY;
import static org.apache.hudi.common.table.HoodieTableConfig.TABLE_METADATA_PARTITIONS;
import static org.apache.hudi.common.util.StringUtils.EMPTY_STRING;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.deleteMetadataPartition;
@@ -654,23 +661,20 @@ public void finalizeWrite(HoodieEngineContext context, String instantTs, List>> invalidFilesByPartition) {
// Now delete partially written files
context.setJobStatus(this.getClass().getSimpleName(), "Delete invalid files generated during the write operation: " + config.getTableName());
- context.map(new ArrayList<>(invalidFilesByPartition.values()), partitionWithFileList -> {
- final FileSystem fileSystem = metaClient.getFs();
- LOG.info("Deleting invalid data files=" + partitionWithFileList);
- if (partitionWithFileList.isEmpty()) {
- return true;
- }
- // Delete
- partitionWithFileList.stream().map(Pair::getValue).forEach(file -> {
- try {
- fileSystem.delete(new Path(file), false);
- } catch (IOException e) {
- throw new HoodieIOException(e.getMessage(), e);
- }
- });
-
- return true;
- }, config.getFinalizeWriteParallelism());
+ context.map(invalidFilesByPartition.values().stream()
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList()),
+ partitionFilePair -> {
+ final FileSystem fileSystem = metaClient.getFs();
+ LOG.info("Deleting invalid data file=" + partitionFilePair);
+ // Delete
+ try {
+ fileSystem.delete(new Path(partitionFilePair.getValue()), false);
+ } catch (IOException e) {
+ throw new HoodieIOException(e.getMessage(), e);
+ }
+ return true;
+ }, config.getFinalizeWriteParallelism());
}
/**
@@ -800,27 +804,22 @@ public TaskContextSupplier getTaskContextSupplier() {
*/
private void validateSchema() throws HoodieUpsertException, HoodieInsertException {
- if (!config.shouldValidateAvroSchema() || getActiveTimeline().getCommitsTimeline().filterCompletedInstants().empty()) {
+ boolean shouldValidate = config.shouldValidateAvroSchema();
+ boolean allowProjection = config.shouldAllowAutoEvolutionColumnDrop();
+ if ((!shouldValidate && allowProjection)
+ || getActiveTimeline().getCommitsTimeline().filterCompletedInstants().empty()) {
// Check not required
return;
}
- Schema tableSchema;
- Schema writerSchema;
- boolean isValid;
try {
TableSchemaResolver schemaResolver = new TableSchemaResolver(getMetaClient());
- writerSchema = HoodieAvroUtils.createHoodieWriteSchema(config.getSchema());
- tableSchema = HoodieAvroUtils.createHoodieWriteSchema(schemaResolver.getTableAvroSchemaWithoutMetadataFields());
- isValid = isSchemaCompatible(tableSchema, writerSchema);
+ Schema writerSchema = HoodieAvroUtils.createHoodieWriteSchema(config.getSchema());
+ Schema tableSchema = HoodieAvroUtils.createHoodieWriteSchema(schemaResolver.getTableAvroSchema(false));
+ AvroSchemaUtils.checkSchemaCompatible(tableSchema, writerSchema, shouldValidate, allowProjection, getDropPartitionColNames());
} catch (Exception e) {
throw new HoodieException("Failed to read schema/check compatibility for base path " + metaClient.getBasePath(), e);
}
-
- if (!isValid) {
- throw new HoodieException("Failed schema compatibility check for writerSchema :" + writerSchema
- + ", table schema :" + tableSchema + ", base path :" + metaClient.getBasePath());
- }
}
public void validateUpsertSchema() throws HoodieUpsertException {
@@ -872,7 +871,8 @@ public HoodieEngineContext getContext() {
* @return instance of {@link HoodieTableMetadataWriter}
*/
public final Option getMetadataWriter(String triggeringInstantTimestamp) {
- return getMetadataWriter(triggeringInstantTimestamp, Option.empty());
+ return getMetadataWriter(
+ triggeringInstantTimestamp, EAGER, Option.empty());
}
/**
@@ -895,6 +895,29 @@ public boolean isTableServiceAction(String actionType, String instantTime) {
}
}
+ /**
+ * Gets the metadata writer for async indexer.
+ *
+ * @param triggeringInstantTimestamp The instant that is triggering this metadata write.
+ * @return An instance of {@link HoodieTableMetadataWriter}.
+ */
+ public Option getIndexingMetadataWriter(String triggeringInstantTimestamp) {
+ return getMetadataWriter(triggeringInstantTimestamp, LAZY, Option.empty());
+ }
+
+ /**
+ * Gets the metadata writer for regular writes.
+ *
+ * @param triggeringInstantTimestamp The instant that is triggering this metadata write.
+ * @param actionMetadata Optional action metadata.
+ * @param Action metadata type.
+ * @return An instance of {@link HoodieTableMetadataWriter}.
+ */
+ public Option