lastCompletedTxnOwnerInstant) {
+ if (!this.currentTxnOwnerInstant.isPresent() || this.currentTxnOwnerInstant.get().equals(callerInstant.get())) {
+ this.currentTxnOwnerInstant = newTxnOwnerInstant;
+ this.lastCompletedTxnOwnerInstant = lastCompletedTxnOwnerInstant;
}
}
public void close() {
- if (supportsOptimisticConcurrency) {
+ if (isOptimisticConcurrencyControlEnabled) {
lockManager.close();
LOG.info("Transaction manager closed");
}
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
new file mode 100644
index 000000000000..cab9d95df4be
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/InProcessLockProvider.java
@@ -0,0 +1,117 @@
+/*
+ * 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.lock;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hudi.common.config.LockConfiguration;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.lock.LockProvider;
+import org.apache.hudi.common.lock.LockState;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.exception.HoodieLockException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * InProcess level lock. This {@link LockProvider} implementation is to
+ * guard table from concurrent operations happening in the local JVM process.
+ *
+ * Note: This Lock provider implementation doesn't allow lock reentrancy.
+ * Attempting to reacquire the lock from the same thread will throw
+ * HoodieLockException. Threads other than the current lock owner, will
+ * block on lock() and return false on tryLock().
+ */
+public class InProcessLockProvider implements LockProvider {
+
+ private static final Logger LOG = LogManager.getLogger(ZookeeperBasedLockProvider.class);
+ private static final ReentrantReadWriteLock LOCK = new ReentrantReadWriteLock();
+ private final long maxWaitTimeMillis;
+
+ public InProcessLockProvider(final LockConfiguration lockConfiguration, final Configuration conf) {
+ TypedProperties typedProperties = lockConfiguration.getConfig();
+ maxWaitTimeMillis = typedProperties.getLong(LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY,
+ LockConfiguration.DEFAULT_ACQUIRE_LOCK_WAIT_TIMEOUT_MS);
+ }
+
+ @Override
+ public void lock() {
+ LOG.info(getLogMessage(LockState.ACQUIRING));
+ if (LOCK.isWriteLockedByCurrentThread()) {
+ throw new HoodieLockException(getLogMessage(LockState.ALREADY_ACQUIRED));
+ }
+ LOCK.writeLock().lock();
+ LOG.info(getLogMessage(LockState.ACQUIRED));
+ }
+
+ @Override
+ public boolean tryLock() {
+ return tryLock(maxWaitTimeMillis, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public boolean tryLock(long time, @NotNull TimeUnit unit) {
+ LOG.info(getLogMessage(LockState.ACQUIRING));
+ if (LOCK.isWriteLockedByCurrentThread()) {
+ throw new HoodieLockException(getLogMessage(LockState.ALREADY_ACQUIRED));
+ }
+
+ boolean isLockAcquired;
+ try {
+ isLockAcquired = LOCK.writeLock().tryLock(time, unit);
+ } catch (InterruptedException e) {
+ throw new HoodieLockException(getLogMessage(LockState.FAILED_TO_ACQUIRE));
+ }
+
+ LOG.info(getLogMessage(isLockAcquired ? LockState.ACQUIRED : LockState.FAILED_TO_ACQUIRE));
+ return isLockAcquired;
+ }
+
+ @Override
+ public void unlock() {
+ LOG.info(getLogMessage(LockState.RELEASING));
+ try {
+ LOCK.writeLock().unlock();
+ } catch (Exception e) {
+ throw new HoodieLockException(getLogMessage(LockState.FAILED_TO_RELEASE), e);
+ }
+ LOG.info(getLogMessage(LockState.RELEASED));
+ }
+
+ @Override
+ public ReentrantReadWriteLock getLock() {
+ return LOCK;
+ }
+
+ @Override
+ public void close() {
+ if (LOCK.isWriteLockedByCurrentThread()) {
+ LOCK.writeLock().unlock();
+ }
+ }
+
+ private String getLogMessage(LockState state) {
+ return StringUtils.join("Thread ", String.valueOf(Thread.currentThread().getName()), " ",
+ state.name(), " in-process lock.");
+ }
+}
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 33519b9adc0d..976205f3592e 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
@@ -20,13 +20,10 @@
import java.io.Serializable;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hudi.common.config.LockConfiguration;
import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.lock.LockProvider;
-import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieLockException;
@@ -46,11 +43,8 @@ public class LockManager implements Serializable, AutoCloseable {
private final LockConfiguration lockConfiguration;
private final SerializableConfiguration hadoopConf;
private volatile LockProvider lockProvider;
- // Holds the latest completed write instant to know which ones to check conflict against
- private final AtomicReference> latestCompletedWriteInstant;
public LockManager(HoodieWriteConfig writeConfig, FileSystem fs) {
- this.latestCompletedWriteInstant = new AtomicReference<>(Option.empty());
this.writeConfig = writeConfig;
this.hadoopConf = new SerializableConfiguration(fs.getConf());
this.lockConfiguration = new LockConfiguration(writeConfig.getProps());
@@ -72,7 +66,7 @@ public void lock() {
LOG.info("Retrying to acquire lock...");
Thread.sleep(waitTimeInMs);
retryCount++;
- } catch (InterruptedException e) {
+ } catch (HoodieLockException | InterruptedException e) {
if (retryCount >= retries) {
throw new HoodieLockException("Unable to acquire lock, lock object ", e);
}
@@ -100,22 +94,6 @@ public synchronized LockProvider getLockProvider() {
return lockProvider;
}
- public void setLatestCompletedWriteInstant(Option instant) {
- this.latestCompletedWriteInstant.set(instant);
- }
-
- public void compareAndSetLatestCompletedWriteInstant(Option expected, Option newValue) {
- this.latestCompletedWriteInstant.compareAndSet(expected, newValue);
- }
-
- public AtomicReference> getLatestCompletedWriteInstant() {
- return latestCompletedWriteInstant;
- }
-
- public void resetLatestCompletedWriteInstant() {
- this.latestCompletedWriteInstant.set(Option.empty());
- }
-
@Override
public void close() {
closeQuietly();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
index 8e59478da05f..ed2ea457764f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
@@ -18,11 +18,8 @@
package org.apache.hudi.client.utils;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import org.apache.hudi.client.transaction.ConflictResolutionStrategy;
import org.apache.hudi.client.transaction.ConcurrentOperation;
+import org.apache.hudi.client.transaction.ConflictResolutionStrategy;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -34,9 +31,12 @@
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieWriteConflictException;
import org.apache.hudi.table.HoodieTable;
+
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
+
import java.io.IOException;
+import java.util.Map;
import java.util.stream.Stream;
public class TransactionUtils {
@@ -45,6 +45,7 @@ public class TransactionUtils {
/**
* Resolve any write conflicts when committing data.
+ *
* @param table
* @param currentTxnOwnerInstant
* @param thisCommitMetadata
@@ -54,11 +55,11 @@ public class TransactionUtils {
* @throws HoodieWriteConflictException
*/
public static Option resolveWriteConflictIfAny(
- final HoodieTable table,
- final Option currentTxnOwnerInstant,
- final Option thisCommitMetadata,
- final HoodieWriteConfig config,
- Option lastCompletedTxnOwnerInstant) throws HoodieWriteConflictException {
+ final HoodieTable table,
+ final Option currentTxnOwnerInstant,
+ final Option thisCommitMetadata,
+ final HoodieWriteConfig config,
+ Option lastCompletedTxnOwnerInstant) throws HoodieWriteConflictException {
if (config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()) {
ConflictResolutionStrategy resolutionStrategy = config.getWriteConflictResolutionStrategy();
Stream instantStream = resolutionStrategy.getCandidateInstants(table.getActiveTimeline(), currentTxnOwnerInstant.get(), lastCompletedTxnOwnerInstant);
@@ -68,7 +69,7 @@ public static Option resolveWriteConflictIfAny(
ConcurrentOperation otherOperation = new ConcurrentOperation(instant, table.getMetaClient());
if (resolutionStrategy.hasConflict(thisOperation, otherOperation)) {
LOG.info("Conflict encountered between current instant = " + thisOperation + " and instant = "
- + otherOperation + ", attempting to resolve it...");
+ + otherOperation + ", attempting to resolve it...");
resolutionStrategy.resolveConflict(table, thisOperation, otherOperation);
}
} catch (IOException io) {
@@ -77,10 +78,6 @@ public static Option resolveWriteConflictIfAny(
});
LOG.info("Successfully resolved conflicts, if any");
- if (config.mergeDeltastreamerStateFromPreviousCommit()) {
- mergeCheckpointStateFromPreviousCommit(table.getMetaClient(), thisOperation.getCommitMetadataOption());
- }
-
return thisOperation.getCommitMetadataOption();
}
return thisCommitMetadata;
@@ -88,6 +85,7 @@ public static Option resolveWriteConflictIfAny(
/**
* Get the last completed transaction hoodie instant and {@link HoodieCommitMetadata#getExtraMetadata()}.
+ *
* @param metaClient
* @return
*/
@@ -117,30 +115,4 @@ public static Option>> getLastCompletedT
throw new HoodieIOException("Unable to read metadata for instant " + hoodieInstantOption.get(), io);
}
}
-
- protected static void mergeCheckpointStateFromPreviousCommit(HoodieTableMetaClient metaClient, Option thisMetadata) {
- overrideWithLatestCommitMetadata(metaClient, thisMetadata, Collections.singletonList(HoodieWriteConfig.DELTASTREAMER_CHECKPOINT_KEY));
- }
-
- /**
- * Generic method allowing us to override the current metadata with the metadata from
- * the latest instant for the specified key prefixes.
- * @param metaClient
- * @param thisMetadata
- * @param keyPrefixes The key prefixes to merge from the previous commit
- */
- private static void overrideWithLatestCommitMetadata(HoodieTableMetaClient metaClient,
- Option thisMetadata,
- List keyPrefixes) {
- if (keyPrefixes.size() == 1 && keyPrefixes.get(0).length() < 1) {
- return;
- }
- Option>> lastInstant = getLastCompletedTxnInstantAndMetadata(metaClient);
- if (lastInstant.isPresent() && thisMetadata.isPresent()) {
- Stream lastCommitMetadataKeys = lastInstant.get().getRight().keySet().stream();
- keyPrefixes.stream().forEach(keyPrefix -> lastCommitMetadataKeys
- .filter(key -> key.startsWith(keyPrefix))
- .forEach(key -> thisMetadata.get().getExtraMetadata().put(key, lastInstant.get().getRight().get(key))));
- }
- }
}
\ No newline at end of file
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieClusteringConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieClusteringConfig.java
index 73c66d8038e5..9486ad43de60 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieClusteringConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieClusteringConfig.java
@@ -22,12 +22,18 @@
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.TypedProperties;
+import org.apache.hudi.common.engine.EngineType;
+import org.apache.hudi.common.util.TypeUtils;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.table.action.cluster.ClusteringPlanPartitionFilterMode;
+import javax.annotation.Nonnull;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
-import java.util.Locale;
+import java.util.Map;
import java.util.Properties;
/**
@@ -41,6 +47,16 @@ public class HoodieClusteringConfig extends HoodieConfig {
// Any strategy specific params can be saved with this prefix
public static final String CLUSTERING_STRATEGY_PARAM_PREFIX = "hoodie.clustering.plan.strategy.";
+ public static final String SPARK_SIZED_BASED_CLUSTERING_PLAN_STRATEGY =
+ "org.apache.hudi.client.clustering.plan.strategy.SparkSizeBasedClusteringPlanStrategy";
+ public static final String JAVA_SIZED_BASED_CLUSTERING_PLAN_STRATEGY =
+ "org.apache.hudi.client.clustering.plan.strategy.JavaSizeBasedClusteringPlanStrategy";
+ public static final String SPARK_SORT_AND_SIZE_EXECUTION_STRATEGY =
+ "org.apache.hudi.client.clustering.run.strategy.SparkSortAndSizeExecutionStrategy";
+ public static final String JAVA_SORT_AND_SIZE_EXECUTION_STRATEGY =
+ "org.apache.hudi.client.clustering.run.strategy.JavaSortAndSizeExecutionStrategy";
+ public static final String PLAN_PARTITION_FILTER_MODE =
+ "hoodie.clustering.plan.partition.filter.mode";
// Any Space-filling curves optimize(z-order/hilbert) params can be saved with this prefix
public static final String LAYOUT_OPTIMIZE_PARAM_PREFIX = "hoodie.layout.optimize.";
@@ -51,6 +67,20 @@ public class HoodieClusteringConfig extends HoodieConfig {
.sinceVersion("0.7.0")
.withDocumentation("Number of partitions to list to create ClusteringPlan");
+ public static final ConfigProperty PARTITION_FILTER_BEGIN_PARTITION = ConfigProperty
+ .key(CLUSTERING_STRATEGY_PARAM_PREFIX + "cluster.begin.partition")
+ .noDefaultValue()
+ .sinceVersion("0.11.0")
+ .withDocumentation("Begin partition used to filter partition (inclusive), only effective when the filter mode '"
+ + PLAN_PARTITION_FILTER_MODE + "' is " + ClusteringPlanPartitionFilterMode.SELECTED_PARTITIONS.name());
+
+ public static final ConfigProperty PARTITION_FILTER_END_PARTITION = ConfigProperty
+ .key(CLUSTERING_STRATEGY_PARAM_PREFIX + "cluster.end.partition")
+ .noDefaultValue()
+ .sinceVersion("0.11.0")
+ .withDocumentation("End partition used to filter partition (inclusive), only effective when the filter mode '"
+ + PLAN_PARTITION_FILTER_MODE + "' is " + ClusteringPlanPartitionFilterMode.SELECTED_PARTITIONS.name());
+
public static final ConfigProperty PLAN_STRATEGY_SMALL_FILE_LIMIT = ConfigProperty
.key(CLUSTERING_STRATEGY_PARAM_PREFIX + "small.file.limit")
.defaultValue(String.valueOf(600 * 1024 * 1024L))
@@ -59,7 +89,7 @@ public class HoodieClusteringConfig extends HoodieConfig {
public static final ConfigProperty PLAN_STRATEGY_CLASS_NAME = ConfigProperty
.key("hoodie.clustering.plan.strategy.class")
- .defaultValue("org.apache.hudi.client.clustering.plan.strategy.SparkSizeBasedClusteringPlanStrategy")
+ .defaultValue(SPARK_SIZED_BASED_CLUSTERING_PLAN_STRATEGY)
.sinceVersion("0.7.0")
.withDocumentation("Config to provide a strategy class (subclass of ClusteringPlanStrategy) to create clustering plan "
+ "i.e select what file groups are being clustered. Default strategy, looks at the clustering small file size limit (determined by "
@@ -67,7 +97,7 @@ public class HoodieClusteringConfig extends HoodieConfig {
public static final ConfigProperty EXECUTION_STRATEGY_CLASS_NAME = ConfigProperty
.key("hoodie.clustering.execution.strategy.class")
- .defaultValue("org.apache.hudi.client.clustering.run.strategy.SparkSortAndSizeExecutionStrategy")
+ .defaultValue(SPARK_SORT_AND_SIZE_EXECUTION_STRATEGY)
.sinceVersion("0.7.0")
.withDocumentation("Config to provide a strategy class (subclass of RunClusteringStrategy) to define how the "
+ " clustering plan is executed. By default, we sort the file groups in th plan by the specified columns, while "
@@ -97,6 +127,17 @@ public class HoodieClusteringConfig extends HoodieConfig {
.sinceVersion("0.9.0")
.withDocumentation("Number of partitions to skip from latest when choosing partitions to create ClusteringPlan");
+ public static final ConfigProperty PLAN_PARTITION_FILTER_MODE_NAME = ConfigProperty
+ .key(PLAN_PARTITION_FILTER_MODE)
+ .defaultValue(ClusteringPlanPartitionFilterMode.NONE)
+ .sinceVersion("0.11.0")
+ .withDocumentation("Partition filter mode used in the creation of clustering plan. Available values are - "
+ + "NONE: do not filter table partition and thus the clustering plan will include all partitions that have clustering candidate."
+ + "RECENT_DAYS: keep a continuous range of partitions, worked together with configs '" + DAYBASED_LOOKBACK_PARTITIONS.key() + "' and '"
+ + PLAN_STRATEGY_SKIP_PARTITIONS_FROM_LATEST.key() + "."
+ + "SELECTED_PARTITIONS: keep partitions that are in the specified range ['" + PARTITION_FILTER_BEGIN_PARTITION.key() + "', '"
+ + PARTITION_FILTER_END_PARTITION.key() + "'].");
+
public static final ConfigProperty PLAN_STRATEGY_MAX_BYTES_PER_OUTPUT_FILEGROUP = ConfigProperty
.key(CLUSTERING_STRATEGY_PARAM_PREFIX + "max.bytes.per.group")
.defaultValue(String.valueOf(2 * 1024 * 1024 * 1024L))
@@ -142,6 +183,17 @@ public class HoodieClusteringConfig extends HoodieConfig {
.sinceVersion("0.9.0")
.withDocumentation("When rewriting data, preserves existing hoodie_commit_time");
+ /**
+ * Using space-filling curves to optimize the layout of table to boost query performance.
+ * The table data which sorted by space-filling curve has better aggregation;
+ * combine with min-max filtering, it can achieve good performance improvement.
+ *
+ * Notice:
+ * when we use this feature, we need specify the sort columns.
+ * The more columns involved in sorting, the worse the aggregation, and the smaller the query performance improvement.
+ * Choose the filter columns which commonly used in query sql as sort columns.
+ * It is recommend that 2 ~ 4 columns participate in sorting.
+ */
public static final ConfigProperty LAYOUT_OPTIMIZE_ENABLE = ConfigProperty
.key(LAYOUT_OPTIMIZE_PARAM_PREFIX + "enable")
.defaultValue(false)
@@ -191,6 +243,15 @@ public class HoodieClusteringConfig extends HoodieConfig {
.sinceVersion("0.10.0")
.withDocumentation("Enable data skipping by collecting statistics once layout optimization is complete.");
+ public static final ConfigProperty ROLLBACK_PENDING_CLUSTERING_ON_CONFLICT = ConfigProperty
+ .key("hoodie.clustering.rollback.pending.replacecommit.on.conflict")
+ .defaultValue(false)
+ .sinceVersion("0.10.0")
+ .withDocumentation("If updates are allowed to file groups pending clustering, then set this config to rollback failed or pending clustering instants. "
+ + "Pending clustering will be rolled back ONLY IF there is conflict between incoming upsert and filegroup to be clustered. "
+ + "Please exercise caution while setting this config, especially when clustering is done very frequently. This could lead to race condition in "
+ + "rare scenarios, for example, when the clustering completes after instants are fetched but before rollback completed.");
+
/**
* @deprecated Use {@link #PLAN_STRATEGY_CLASS_NAME} and its methods instead
*/
@@ -304,11 +365,24 @@ public class HoodieClusteringConfig extends HoodieConfig {
/** @deprecated Use {@link #ASYNC_CLUSTERING_ENABLE} and its methods instead */
@Deprecated
public static final String DEFAULT_ASYNC_CLUSTERING_ENABLE_OPT_VAL = ASYNC_CLUSTERING_ENABLE.defaultValue();
-
+
+ // NOTE: This ctor is required for appropriate deserialization
public HoodieClusteringConfig() {
super();
}
+ public boolean isAsyncClusteringEnabled() {
+ return getBooleanOrDefault(HoodieClusteringConfig.ASYNC_CLUSTERING_ENABLE);
+ }
+
+ public boolean isInlineClusteringEnabled() {
+ return getBooleanOrDefault(HoodieClusteringConfig.INLINE_CLUSTERING);
+ }
+
+ public static HoodieClusteringConfig from(TypedProperties props) {
+ return HoodieClusteringConfig.newBuilder().fromProperties(props).build();
+ }
+
public static Builder newBuilder() {
return new Builder();
}
@@ -316,6 +390,12 @@ public static Builder newBuilder() {
public static class Builder {
private final HoodieClusteringConfig clusteringConfig = new HoodieClusteringConfig();
+ private EngineType engineType = EngineType.SPARK;
+
+ public Builder withEngineType(EngineType engineType) {
+ this.engineType = engineType;
+ return this;
+ }
public Builder fromFile(File propertiesFile) throws IOException {
try (FileReader reader = new FileReader(propertiesFile)) {
@@ -329,6 +409,11 @@ public Builder withClusteringPlanStrategyClass(String clusteringStrategyClass) {
return this;
}
+ public Builder withClusteringPlanPartitionFilterMode(ClusteringPlanPartitionFilterMode mode) {
+ clusteringConfig.setValue(PLAN_PARTITION_FILTER_MODE_NAME.key(), mode.toString());
+ return this;
+ }
+
public Builder withClusteringExecutionStrategyClass(String runClusteringStrategyClass) {
clusteringConfig.setValue(EXECUTION_STRATEGY_CLASS_NAME, runClusteringStrategyClass);
return this;
@@ -344,6 +429,16 @@ public Builder withClusteringSkipPartitionsFromLatest(int clusteringSkipPartitio
return this;
}
+ public Builder withClusteringPartitionFilterBeginPartition(String begin) {
+ clusteringConfig.setValue(PARTITION_FILTER_BEGIN_PARTITION, begin);
+ return this;
+ }
+
+ public Builder withClusteringPartitionFilterEndPartition(String end) {
+ clusteringConfig.setValue(PARTITION_FILTER_END_PARTITION, end);
+ return this;
+ }
+
public Builder withClusteringPlanSmallFileLimit(long clusteringSmallFileLimit) {
clusteringConfig.setValue(PLAN_STRATEGY_SMALL_FILE_LIMIT, String.valueOf(clusteringSmallFileLimit));
return this;
@@ -385,6 +480,7 @@ public Builder withAsyncClusteringMaxCommits(int numCommits) {
}
public Builder fromProperties(Properties props) {
+ // TODO this should cherry-pick only clustering properties
this.clusteringConfig.getProps().putAll(props);
return this;
}
@@ -404,6 +500,11 @@ public Builder withPreserveHoodieCommitMetadata(Boolean preserveHoodieCommitMeta
return this;
}
+ public Builder withRollbackPendingClustering(Boolean rollbackPendingClustering) {
+ clusteringConfig.setValue(ROLLBACK_PENDING_CLUSTERING_ON_CONFLICT, String.valueOf(rollbackPendingClustering));
+ return this;
+ }
+
public Builder withSpaceFillingCurveDataOptimizeEnable(Boolean enable) {
clusteringConfig.setValue(LAYOUT_OPTIMIZE_ENABLE, String.valueOf(enable));
return this;
@@ -430,17 +531,49 @@ public Builder withDataOptimizeDataSkippingEnable(boolean dataSkipping) {
}
public HoodieClusteringConfig build() {
+ clusteringConfig.setDefaultValue(
+ PLAN_STRATEGY_CLASS_NAME, getDefaultPlanStrategyClassName(engineType));
+ clusteringConfig.setDefaultValue(
+ EXECUTION_STRATEGY_CLASS_NAME, getDefaultExecutionStrategyClassName(engineType));
clusteringConfig.setDefaults(HoodieClusteringConfig.class.getName());
return clusteringConfig;
}
+
+ private String getDefaultPlanStrategyClassName(EngineType engineType) {
+ switch (engineType) {
+ case SPARK:
+ return SPARK_SIZED_BASED_CLUSTERING_PLAN_STRATEGY;
+ case FLINK:
+ case JAVA:
+ return JAVA_SIZED_BASED_CLUSTERING_PLAN_STRATEGY;
+ default:
+ throw new HoodieNotSupportedException("Unsupported engine " + engineType);
+ }
+ }
+
+ private String getDefaultExecutionStrategyClassName(EngineType engineType) {
+ switch (engineType) {
+ case SPARK:
+ return SPARK_SORT_AND_SIZE_EXECUTION_STRATEGY;
+ case FLINK:
+ case JAVA:
+ return JAVA_SORT_AND_SIZE_EXECUTION_STRATEGY;
+ default:
+ throw new HoodieNotSupportedException("Unsupported engine " + engineType);
+ }
+ }
}
/**
- * strategy types for build z-ordering/space-filling curves.
+ * Type of a strategy for building Z-order/Hilbert space-filling curves.
*/
public enum BuildCurveStrategyType {
DIRECT("direct"),
SAMPLE("sample");
+
+ private static final Map VALUE_TO_ENUM_MAP =
+ TypeUtils.getValueToEnumMap(BuildCurveStrategyType.class, e -> e.value);
+
private final String value;
BuildCurveStrategyType(String value) {
@@ -448,14 +581,39 @@ public enum BuildCurveStrategyType {
}
public static BuildCurveStrategyType fromValue(String value) {
- switch (value.toLowerCase(Locale.ROOT)) {
- case "direct":
- return DIRECT;
- case "sample":
- return SAMPLE;
- default:
- throw new HoodieException("Invalid value of Type.");
+ BuildCurveStrategyType enumValue = VALUE_TO_ENUM_MAP.get(value);
+ if (enumValue == null) {
+ throw new HoodieException(String.format("Invalid value (%s)", value));
}
+
+ return enumValue;
+ }
+ }
+
+ /**
+ * Layout optimization strategies such as Z-order/Hilbert space-curves, etc
+ */
+ public enum LayoutOptimizationStrategy {
+ ZORDER("z-order"),
+ HILBERT("hilbert");
+
+ private static final Map VALUE_TO_ENUM_MAP =
+ TypeUtils.getValueToEnumMap(LayoutOptimizationStrategy.class, e -> e.value);
+
+ private final String value;
+
+ LayoutOptimizationStrategy(String value) {
+ this.value = value;
+ }
+
+ @Nonnull
+ public static LayoutOptimizationStrategy fromValue(String value) {
+ LayoutOptimizationStrategy enumValue = VALUE_TO_ENUM_MAP.get(value);
+ if (enumValue == null) {
+ throw new HoodieException(String.format("Invalid value (%s)", value));
+ }
+
+ return enumValue;
}
}
}
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 ce74aad6b098..640f0cb1826a 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
@@ -57,6 +57,13 @@ public class HoodieCompactionConfig extends HoodieConfig {
+ " to delete older file slices. It's recommended to enable this, to ensure metadata and data storage"
+ " growth is bounded.");
+ public static final ConfigProperty AUTO_ARCHIVE = ConfigProperty
+ .key("hoodie.archive.automatic")
+ .defaultValue("true")
+ .withDocumentation("When enabled, the archival table service is invoked immediately after each commit,"
+ + " to archive commits if we cross a maximum value of commits."
+ + " It's recommended to enable this, to ensure number of active commits is bounded.");
+
public static final ConfigProperty ASYNC_CLEAN = ConfigProperty
.key("hoodie.clean.async")
.defaultValue("false")
@@ -119,6 +126,11 @@ public class HoodieCompactionConfig extends HoodieConfig {
+ " keep the metadata overhead constant, even as the table size grows."
+ "This config controls the maximum number of instants to retain in the active timeline. ");
+ public static final ConfigProperty DELETE_ARCHIVED_INSTANT_PARALLELISM_VALUE = ConfigProperty
+ .key("hoodie.archive.delete.parallelism")
+ .defaultValue(100)
+ .withDocumentation("Parallelism for deleting archived hoodie commits.");
+
public static final ConfigProperty MIN_COMMITS_TO_KEEP = ConfigProperty
.key("hoodie.keep.min.commits")
.defaultValue("20")
@@ -166,6 +178,12 @@ public class HoodieCompactionConfig extends HoodieConfig {
.withDocumentation("Amount of MBs to spend during compaction run for the LogFileSizeBasedCompactionStrategy. "
+ "This value helps bound ingestion latency while compaction is run inline mode.");
+ public static final ConfigProperty COMPACTION_LOG_FILE_SIZE_THRESHOLD = ConfigProperty
+ .key("hoodie.compaction.logfile.size.threshold")
+ .defaultValue(0L)
+ .withDocumentation("Only if the log file size is greater than the threshold in bytes,"
+ + " the file group will be compacted.");
+
public static final ConfigProperty COMPACTION_STRATEGY = ConfigProperty
.key("hoodie.compaction.strategy")
.defaultValue(LogFileSizeBasedCompactionStrategy.class.getName())
@@ -482,6 +500,11 @@ public Builder withAsyncClean(Boolean asyncClean) {
return this;
}
+ public Builder withAutoArchive(Boolean autoArchive) {
+ compactionConfig.setValue(AUTO_ARCHIVE, String.valueOf(autoArchive));
+ return this;
+ }
+
public Builder withIncrementalCleaningMode(Boolean incrementalCleaningMode) {
compactionConfig.setValue(CLEANER_INCREMENTAL_MODE_ENABLE, String.valueOf(incrementalCleaningMode));
return this;
@@ -568,6 +591,11 @@ public Builder withMaxNumDeltaCommitsBeforeCompaction(int maxNumDeltaCommitsBefo
return this;
}
+ public Builder withArchiveDeleteParallelism(int archiveDeleteParallelism) {
+ compactionConfig.setValue(DELETE_ARCHIVED_INSTANT_PARALLELISM_VALUE, String.valueOf(archiveDeleteParallelism));
+ return this;
+ }
+
public Builder withMaxDeltaSecondsBeforeCompaction(int maxDeltaSecondsBeforeCompaction) {
compactionConfig.setValue(INLINE_COMPACT_TIME_DELTA_SECONDS, String.valueOf(maxDeltaSecondsBeforeCompaction));
return this;
@@ -588,6 +616,11 @@ public Builder withTargetPartitionsPerDayBasedCompaction(int targetPartitionsPer
return this;
}
+ public Builder withLogFileSizeThresholdBasedCompaction(long logFileSizeThreshold) {
+ compactionConfig.setValue(COMPACTION_LOG_FILE_SIZE_THRESHOLD, String.valueOf(logFileSizeThreshold));
+ return this;
+ }
+
public Builder withCommitsArchivalBatchSize(int batchSize) {
compactionConfig.setValue(COMMITS_ARCHIVAL_BATCH_SIZE, String.valueOf(batchSize));
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 561460777ff6..3d7e3a7941da 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
@@ -76,9 +76,9 @@ public class HoodieHBaseIndexConfig extends HoodieConfig {
.withDocumentation("Property to set which implementation of HBase QPS resource allocator to be used, which"
+ "controls the batching rate dynamically.");
- public static final ConfigProperty PUT_BATCH_SIZE_AUTO_COMPUTE = ConfigProperty
+ public static final ConfigProperty PUT_BATCH_SIZE_AUTO_COMPUTE = ConfigProperty
.key("hoodie.index.hbase.put.batch.size.autocompute")
- .defaultValue("false")
+ .defaultValue(false)
.withDocumentation("Property to set to enable auto computation of put batch size");
public static final ConfigProperty QPS_FRACTION = ConfigProperty
@@ -206,7 +206,7 @@ public class HoodieHBaseIndexConfig extends HoodieConfig {
* @deprecated Use {@link #PUT_BATCH_SIZE_AUTO_COMPUTE} and its methods instead
*/
@Deprecated
- public static final String DEFAULT_HBASE_PUT_BATCH_SIZE_AUTO_COMPUTE = PUT_BATCH_SIZE_AUTO_COMPUTE.defaultValue();
+ public static final Boolean DEFAULT_HBASE_PUT_BATCH_SIZE_AUTO_COMPUTE = PUT_BATCH_SIZE_AUTO_COMPUTE.defaultValue();
/**
* @deprecated Use {@link #MAX_QPS_FRACTION} and its methods instead
*/
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 4f9b9fcd50e8..e156310c736b 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
@@ -24,15 +24,20 @@
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.engine.EngineType;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.exception.HoodieIndexException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import javax.annotation.concurrent.Immutable;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
+import java.util.Arrays;
import java.util.Properties;
+import java.util.stream.Collectors;
import static org.apache.hudi.config.HoodieHBaseIndexConfig.GET_BATCH_SIZE;
import static org.apache.hudi.config.HoodieHBaseIndexConfig.PUT_BATCH_SIZE;
@@ -54,7 +59,7 @@ public class HoodieIndexConfig extends HoodieConfig {
.key("hoodie.index.type")
.noDefaultValue()
.withDocumentation("Type of index to use. Default is Bloom filter. "
- + "Possible options are [BLOOM | GLOBAL_BLOOM |SIMPLE | GLOBAL_SIMPLE | INMEMORY | HBASE]. "
+ + "Possible options are [BLOOM | GLOBAL_BLOOM |SIMPLE | GLOBAL_SIMPLE | INMEMORY | HBASE | BUCKET]. "
+ "Bloom filters removes the dependency on a external system "
+ "and is stored in the footer of the Parquet Data Files");
@@ -200,6 +205,30 @@ public class HoodieIndexConfig extends HoodieConfig {
.defaultValue("true")
.withDocumentation("Similar to " + BLOOM_INDEX_UPDATE_PARTITION_PATH_ENABLE + ", but for simple index.");
+ /**
+ * ***** Bucket Index Configs *****
+ * Bucket Index is targeted to locate the record fast by hash in big data scenarios.
+ * The current implementation is a basic version, so there are some constraints:
+ * 1. Unsupported operation: bulk insert, cluster and so on.
+ * 2. Bucket num change requires rewriting the partition.
+ * 3. Predict the table size and future data growth well to set a reasonable bucket num.
+ * 4. A bucket size is recommended less than 3GB and avoid bing too small.
+ * more details and progress see [HUDI-3039].
+ */
+ // Bucket num equals file groups num in each partition.
+ // Bucket num can be set according to partition size and file group size.
+ public static final ConfigProperty BUCKET_INDEX_NUM_BUCKETS = ConfigProperty
+ .key("hoodie.bucket.index.num.buckets")
+ .defaultValue(256)
+ .withDocumentation("Only applies if index type is BUCKET_INDEX. Determine the number of buckets in the hudi table, "
+ + "and each partition is divided to N buckets.");
+
+ public static final ConfigProperty BUCKET_INDEX_HASH_FIELD = ConfigProperty
+ .key("hoodie.bucket.index.hash.field")
+ .noDefaultValue()
+ .withDocumentation("Index key. It is used to index the record and find its file group. "
+ + "If not set, use record key field as default");
+
/**
* Deprecated configs. These are now part of {@link HoodieHBaseIndexConfig}.
*/
@@ -516,12 +545,23 @@ public Builder withEngineType(EngineType engineType) {
return this;
}
+ public Builder withBucketNum(String bucketNum) {
+ hoodieIndexConfig.setValue(BUCKET_INDEX_NUM_BUCKETS, bucketNum);
+ return this;
+ }
+
+ public Builder withIndexKeyField(String keyField) {
+ hoodieIndexConfig.setValue(BUCKET_INDEX_HASH_FIELD, keyField);
+ return this;
+ }
+
public HoodieIndexConfig build() {
hoodieIndexConfig.setDefaultValue(INDEX_TYPE, getDefaultIndexType(engineType));
hoodieIndexConfig.setDefaults(HoodieIndexConfig.class.getName());
// Throws IllegalArgumentException if the value set is not a known Hoodie Index Type
HoodieIndex.IndexType.valueOf(hoodieIndexConfig.getString(INDEX_TYPE));
+ validateBucketIndexConfig();
return hoodieIndexConfig;
}
@@ -540,5 +580,27 @@ private String getDefaultIndexType(EngineType engineType) {
public EngineType getEngineType() {
return engineType;
}
+
+ private void validateBucketIndexConfig() {
+ if (hoodieIndexConfig.getString(INDEX_TYPE).equalsIgnoreCase(HoodieIndex.IndexType.BUCKET.toString())) {
+ // check the bucket index hash field
+ if (StringUtils.isNullOrEmpty(hoodieIndexConfig.getString(BUCKET_INDEX_HASH_FIELD))) {
+ hoodieIndexConfig.setValue(BUCKET_INDEX_HASH_FIELD,
+ hoodieIndexConfig.getStringOrDefault(KeyGeneratorOptions.RECORDKEY_FIELD_NAME));
+ } else {
+ boolean valid = Arrays
+ .stream(hoodieIndexConfig.getStringOrDefault(KeyGeneratorOptions.RECORDKEY_FIELD_NAME).split(","))
+ .collect(Collectors.toSet())
+ .containsAll(Arrays.asList(hoodieIndexConfig.getString(BUCKET_INDEX_HASH_FIELD).split(",")));
+ if (!valid) {
+ throw new HoodieIndexException("Bucket index key (if configured) must be subset of record key.");
+ }
+ }
+ // check the bucket num
+ if (hoodieIndexConfig.getIntOrDefault(BUCKET_INDEX_NUM_BUCKETS) <= 0) {
+ throw new HoodieIndexException("When using bucket index, hoodie.bucket.index.num.buckets cannot be negative.");
+ }
+ }
+ }
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLayoutConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLayoutConfig.java
new file mode 100644
index 000000000000..183c00c7bc63
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLayoutConfig.java
@@ -0,0 +1,89 @@
+/*
+ * 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.config;
+
+import org.apache.hudi.common.config.ConfigClassProperty;
+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.index.HoodieIndex;
+import org.apache.hudi.table.storage.HoodieStorageLayout;
+
+import javax.annotation.concurrent.Immutable;
+import java.util.Properties;
+
+/**
+ * Storage layout related config.
+ */
+@Immutable
+@ConfigClassProperty(name = "Layout Configs",
+ groupName = ConfigGroups.Names.WRITE_CLIENT,
+ description = "Configurations that control storage layout and data distribution, "
+ + "which defines how the files are organized within a table.")
+public class HoodieLayoutConfig extends HoodieConfig {
+
+ public static final ConfigProperty LAYOUT_TYPE = ConfigProperty
+ .key("hoodie.storage.layout.type")
+ .defaultValue("DEFAULT")
+ .withDocumentation("Type of storage layout. Possible options are [DEFAULT | BUCKET]");
+
+ public static final ConfigProperty LAYOUT_PARTITIONER_CLASS_NAME = ConfigProperty
+ .key("hoodie.storage.layout.partitioner.class")
+ .noDefaultValue()
+ .withDocumentation("Partitioner class, it is used to distribute data in a specific way.");
+
+ private HoodieLayoutConfig() {
+ super();
+ }
+
+ public static HoodieLayoutConfig.Builder newBuilder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+ public HoodieLayoutConfig layoutConfig = new HoodieLayoutConfig();
+
+ public Builder fromProperties(Properties props) {
+ this.layoutConfig.getProps().putAll(props);
+ return this;
+ }
+
+ public Builder withLayoutType(String type) {
+ layoutConfig.setValue(LAYOUT_TYPE, type);
+ return this;
+ }
+
+ public Builder withLayoutPartitioner(String partitionerClass) {
+ layoutConfig.setValue(LAYOUT_PARTITIONER_CLASS_NAME, partitionerClass);
+ return this;
+ }
+
+ public HoodieLayoutConfig build() {
+ setDefault();
+ return layoutConfig;
+ }
+
+ private void setDefault() {
+ if (layoutConfig.getString(HoodieIndexConfig.INDEX_TYPE.key()).equals(HoodieIndex.IndexType.BUCKET.name())) {
+ layoutConfig.setDefaultValue(LAYOUT_TYPE, HoodieStorageLayout.LayoutType.BUCKET.name());
+ }
+ layoutConfig.setDefaultValue(LAYOUT_TYPE, LAYOUT_TYPE.defaultValue());
+ }
+ }
+}
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 fd7b310a57d6..723c9f3cbff4 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
@@ -25,6 +25,7 @@
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.lock.LockProvider;
+import org.apache.hudi.common.util.Option;
import java.io.File;
import java.io.FileReader;
@@ -53,7 +54,6 @@
import static org.apache.hudi.common.config.LockConfiguration.ZK_PORT_PROP_KEY;
import static org.apache.hudi.common.config.LockConfiguration.ZK_SESSION_TIMEOUT_MS_PROP_KEY;
-
/**
* Hoodie Configs for Locks.
*/
@@ -160,9 +160,11 @@ public class HoodieLockConfig extends HoodieConfig {
public static final ConfigProperty ZK_LOCK_KEY = ConfigProperty
.key(ZK_LOCK_KEY_PROP_KEY)
.noDefaultValue()
+ .withInferFunction(p -> Option.ofNullable(p.getStringOrDefault(HoodieWriteConfig.TBL_NAME, null)))
.sinceVersion("0.8.0")
.withDocumentation("Key name under base_path at which to create a ZNode and acquire lock. "
- + "Final path on zk will look like base_path/lock_key. We recommend setting this to the table name");
+ + "Final path on zk will look like base_path/lock_key. If this parameter is not set, we would "
+ + "set it as the table name");
// Pluggable type of lock provider
public static final ConfigProperty LOCK_PROVIDER_CLASS_NAME = ConfigProperty
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 e167b12c8204..d8e4bd3b2430 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
@@ -45,6 +45,7 @@
import org.apache.hudi.config.metrics.HoodieMetricsGraphiteConfig;
import org.apache.hudi.config.metrics.HoodieMetricsJmxConfig;
import org.apache.hudi.config.metrics.HoodieMetricsPrometheusConfig;
+import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
@@ -52,10 +53,12 @@
import org.apache.hudi.metrics.MetricsReporterType;
import org.apache.hudi.metrics.datadog.DatadogHttpClient.ApiSite;
import org.apache.hudi.table.RandomFileIdPrefixProvider;
+import org.apache.hudi.table.action.cluster.ClusteringPlanPartitionFilterMode;
import org.apache.hudi.table.action.compact.CompactionTriggerStrategy;
import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hudi.table.storage.HoodieStorageLayout;
import org.apache.orc.CompressionKind;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
@@ -120,9 +123,9 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty ROLLBACK_USING_MARKERS_ENABLE = ConfigProperty
.key("hoodie.rollback.using.markers")
- .defaultValue("false")
+ .defaultValue("true")
.withDocumentation("Enables a more efficient mechanism for rollbacks based on the marker files generated "
- + "during the writes. Turned off by default.");
+ + "during the writes. Turned on by default.");
public static final ConfigProperty TIMELINE_LAYOUT_VERSION_NUM = ConfigProperty
.key("hoodie.timeline.layout.version")
@@ -248,14 +251,17 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty MARKERS_TYPE = ConfigProperty
.key("hoodie.write.markers.type")
- .defaultValue(MarkerType.DIRECT.toString())
+ .defaultValue(MarkerType.TIMELINE_SERVER_BASED.toString())
.sinceVersion("0.9.0")
.withDocumentation("Marker type to use. Two modes are supported: "
+ "- DIRECT: individual marker file corresponding to each data file is directly "
+ "created by the writer. "
+ "- TIMELINE_SERVER_BASED: marker operations are all handled at the timeline service "
+ "which serves as a proxy. New marker entries are batch processed and stored "
- + "in a limited number of underlying files for efficiency.");
+ + "in a limited number of underlying files for efficiency. If HDFS is used or "
+ + "timeline server is disabled, DIRECT markers are used as fallback even if this "
+ + "is configure. For Spark structured streaming, this configuration does not "
+ + "take effect, i.e., DIRECT markers are always used for Spark structured streaming.");
public static final ConfigProperty MARKERS_TIMELINE_SERVER_BASED_BATCH_NUM_THREADS = ConfigProperty
.key("hoodie.markers.timeline_server_based.batch.num_threads")
@@ -354,6 +360,12 @@ public class HoodieWriteConfig extends HoodieConfig {
.withDocumentation("When enabled, we allow duplicate keys even if inserts are routed to merge with an existing file (for ensuring file sizing)."
+ " This is only relevant for insert operation, since upsert, delete operations will ensure unique key constraints are maintained.");
+ public static final ConfigProperty MERGE_SMALL_FILE_GROUP_CANDIDATES_LIMIT = ConfigProperty
+ .key("hoodie.merge.small.file.group.candidates.limit")
+ .defaultValue(1)
+ .withDocumentation("Limits number of file groups, whose base file satisfies small-file limit, to consider for appending records during upsert operation. "
+ + "Only applicable to MOR tables");
+
public static final ConfigProperty CLIENT_HEARTBEAT_INTERVAL_IN_MS = ConfigProperty
.key("hoodie.client.heartbeat.interval_in_ms")
.defaultValue(60 * 1000)
@@ -372,14 +384,6 @@ public class HoodieWriteConfig extends HoodieConfig {
+ "OPTIMISTIC_CONCURRENCY_CONTROL: Multiple writers can operate on the table and exactly one of them succeed "
+ "if a conflict (writes affect the same file group) is detected.");
- public static final ConfigProperty WRITE_CONCURRENCY_MERGE_DELTASTREAMER_STATE = ConfigProperty
- .key("hoodie.write.concurrency.merge.deltastreamer.state")
- .defaultValue(false)
- .withAlternatives("hoodie.write.meta.key.prefixes")
- .withDocumentation("If enabled, this writer will merge Deltastreamer state from the previous checkpoint in order to allow both realtime "
- + "and batch writers to ingest into a single table. This should not be enabled on Deltastreamer writers. Enabling this config means,"
- + "for a spark writer, deltastreamer checkpoint will be copied over from previous commit to the current one.");
-
/**
* Currently the use this to specify the write schema.
*/
@@ -1035,6 +1039,10 @@ public boolean allowDuplicateInserts() {
return getBoolean(MERGE_ALLOW_DUPLICATE_ON_INSERTS_ENABLE);
}
+ public int getSmallFileGroupCandidatesLimit() {
+ return getInt(MERGE_SMALL_FILE_GROUP_CANDIDATES_LIMIT);
+ }
+
public EngineType getEngineType() {
return engineType;
}
@@ -1095,6 +1103,10 @@ public boolean isAutoClean() {
return getBoolean(HoodieCompactionConfig.AUTO_CLEAN);
}
+ public boolean isAutoArchive() {
+ return getBoolean(HoodieCompactionConfig.AUTO_ARCHIVE);
+ }
+
public boolean isAsyncClean() {
return getBoolean(HoodieCompactionConfig.ASYNC_CLEAN);
}
@@ -1127,6 +1139,10 @@ public Long getTargetIOPerCompactionInMB() {
return getLong(HoodieCompactionConfig.TARGET_IO_PER_COMPACTION_IN_MB);
}
+ public Long getCompactionLogFileSizeThreshold() {
+ return getLong(HoodieCompactionConfig.COMPACTION_LOG_FILE_SIZE_THRESHOLD);
+ }
+
public Boolean getCompactionLazyBlockReadEnabled() {
return getBoolean(HoodieCompactionConfig.COMPACTION_LAZY_BLOCK_READ_ENABLE);
}
@@ -1135,6 +1151,10 @@ public Boolean getCompactionReverseLogReadEnabled() {
return getBoolean(HoodieCompactionConfig.COMPACTION_REVERSE_LOG_READ_ENABLE);
}
+ public int getArchiveDeleteParallelism() {
+ return getInt(HoodieCompactionConfig.DELETE_ARCHIVED_INSTANT_PARALLELISM_VALUE);
+ }
+
public boolean inlineClusteringEnabled() {
return getBoolean(HoodieClusteringConfig.INLINE_CLUSTERING);
}
@@ -1152,6 +1172,10 @@ public boolean isClusteringEnabled() {
return inlineClusteringEnabled() || isAsyncClusteringEnabled();
}
+ public boolean isRollbackPendingClustering() {
+ return getBoolean(HoodieClusteringConfig.ROLLBACK_PENDING_CLUSTERING_ON_CONFLICT);
+ }
+
public int getInlineClusterMaxCommits() {
return getInt(HoodieClusteringConfig.INLINE_CLUSTERING_MAX_COMMITS);
}
@@ -1192,6 +1216,19 @@ public String getClusteringPlanStrategyClass() {
return getString(HoodieClusteringConfig.PLAN_STRATEGY_CLASS_NAME);
}
+ public ClusteringPlanPartitionFilterMode getClusteringPlanPartitionFilterMode() {
+ String mode = getString(HoodieClusteringConfig.PLAN_PARTITION_FILTER_MODE_NAME);
+ return ClusteringPlanPartitionFilterMode.valueOf(mode);
+ }
+
+ public String getBeginPartitionForClustering() {
+ return getString(HoodieClusteringConfig.PARTITION_FILTER_BEGIN_PARTITION);
+ }
+
+ public String getEndPartitionForClustering() {
+ return getString(HoodieClusteringConfig.PARTITION_FILTER_END_PARTITION);
+ }
+
public String getClusteringExecutionStrategyClass() {
return getString(HoodieClusteringConfig.EXECUTION_STRATEGY_CLASS_NAME);
}
@@ -1295,8 +1332,8 @@ public int getHbaseIndexPutBatchSize() {
return getInt(HoodieHBaseIndexConfig.PUT_BATCH_SIZE);
}
- public Boolean getHbaseIndexPutBatchSizeAutoCompute() {
- return getBoolean(HoodieHBaseIndexConfig.PUT_BATCH_SIZE_AUTO_COMPUTE);
+ public boolean getHbaseIndexPutBatchSizeAutoCompute() {
+ return getBooleanOrDefault(HoodieHBaseIndexConfig.PUT_BATCH_SIZE_AUTO_COMPUTE);
}
public String getHBaseQPSResourceAllocatorClass() {
@@ -1357,7 +1394,7 @@ public int getHbaseIndexMaxQPSPerRegionServer() {
}
public boolean getHbaseIndexUpdatePartitionPath() {
- return getBoolean(HoodieHBaseIndexConfig.UPDATE_PARTITION_PATH_ENABLE);
+ return getBooleanOrDefault(HoodieHBaseIndexConfig.UPDATE_PARTITION_PATH_ENABLE);
}
public int getBloomIndexParallelism() {
@@ -1404,6 +1441,14 @@ public boolean getGlobalSimpleIndexUpdatePartitionPath() {
return getBoolean(HoodieIndexConfig.SIMPLE_INDEX_UPDATE_PARTITION_PATH_ENABLE);
}
+ public int getBucketIndexNumBuckets() {
+ return getIntOrDefault(HoodieIndexConfig.BUCKET_INDEX_NUM_BUCKETS);
+ }
+
+ public String getBucketIndexHashField() {
+ return getString(HoodieIndexConfig.BUCKET_INDEX_HASH_FIELD);
+ }
+
/**
* storage properties.
*/
@@ -1558,6 +1603,22 @@ public List getDatadogMetricTags() {
HoodieMetricsDatadogConfig.METRIC_TAG_VALUES, ",").split("\\s*,\\s*")).collect(Collectors.toList());
}
+ public int getCloudWatchReportPeriodSeconds() {
+ return getInt(HoodieMetricsCloudWatchConfig.REPORT_PERIOD_SECONDS);
+ }
+
+ public String getCloudWatchMetricPrefix() {
+ return getString(HoodieMetricsCloudWatchConfig.METRIC_PREFIX);
+ }
+
+ public String getCloudWatchMetricNamespace() {
+ return getString(HoodieMetricsCloudWatchConfig.METRIC_NAMESPACE);
+ }
+
+ public int getCloudWatchMaxDatumsPerRequest() {
+ return getInt(HoodieMetricsCloudWatchConfig.MAX_DATUMS_PER_REQUEST);
+ }
+
public String getMetricReporterClassName() {
return getString(HoodieMetricsConfig.METRICS_REPORTER_CLASS_NAME);
}
@@ -1590,6 +1651,10 @@ public boolean getPushGatewayRandomJobNameSuffix() {
return getBoolean(HoodieMetricsPrometheusConfig.PUSHGATEWAY_RANDOM_JOBNAME_SUFFIX);
}
+ public String getMetricReporterMetricsNamePrefix() {
+ return getStringOrDefault(HoodieMetricsConfig.METRICS_REPORTER_PREFIX);
+ }
+
/**
* memory configs.
*/
@@ -1760,10 +1825,6 @@ public WriteConcurrencyMode getWriteConcurrencyMode() {
return WriteConcurrencyMode.fromValue(getString(WRITE_CONCURRENCY_MODE));
}
- public Boolean mergeDeltastreamerStateFromPreviousCommit() {
- return getBoolean(HoodieWriteConfig.WRITE_CONCURRENCY_MERGE_DELTASTREAMER_STATE);
- }
-
public Boolean inlineTableServices() {
return inlineClusteringEnabled() || inlineCompactionEnabled() || isAutoClean();
}
@@ -1796,6 +1857,13 @@ public String getFileIdPrefixProviderClassName() {
return getString(FILEID_PREFIX_PROVIDER_CLASS);
}
+ /**
+ * Layout configs.
+ */
+ public HoodieStorageLayout.LayoutType getLayoutType() {
+ return HoodieStorageLayout.LayoutType.valueOf(getString(HoodieLayoutConfig.LAYOUT_TYPE));
+ }
+
public static class Builder {
protected final HoodieWriteConfig writeConfig = new HoodieWriteConfig();
@@ -1817,6 +1885,7 @@ public static class Builder {
private boolean isPreCommitValidationConfigSet = false;
private boolean isMetricsJmxConfigSet = false;
private boolean isMetricsGraphiteConfigSet = false;
+ private boolean isLayoutConfigSet = false;
public Builder withEngineType(EngineType engineType) {
this.engineType = engineType;
@@ -2047,6 +2116,12 @@ public Builder withCallbackConfig(HoodieWriteCommitCallbackConfig callbackConfig
return this;
}
+ public Builder withLayoutConfig(HoodieLayoutConfig layoutConfig) {
+ writeConfig.getProps().putAll(layoutConfig.getProps());
+ isLayoutConfigSet = true;
+ return this;
+ }
+
public Builder withFinalizeWriteParallelism(int parallelism) {
writeConfig.setValue(FINALIZE_WRITE_PARALLELISM_VALUE, String.valueOf(parallelism));
return this;
@@ -2112,6 +2187,11 @@ public Builder withMergeAllowDuplicateOnInserts(boolean routeInsertsToNewFiles)
return this;
}
+ public Builder withMergeSmallFileGroupCandidatesLimit(int limit) {
+ writeConfig.setValue(MERGE_SMALL_FILE_GROUP_CANDIDATES_LIMIT, String.valueOf(limit));
+ return this;
+ }
+
public Builder withHeartbeatIntervalInMs(Integer heartbeatIntervalInMs) {
writeConfig.setValue(CLIENT_HEARTBEAT_INTERVAL_IN_MS, String.valueOf(heartbeatIntervalInMs));
return this;
@@ -2148,6 +2228,7 @@ public Builder withProperties(Properties properties) {
}
protected void setDefaults() {
+ writeConfig.setDefaultValue(MARKERS_TYPE, getDefaultMarkersType(engineType));
// Check for mandatory properties
writeConfig.setDefaults(HoodieWriteConfig.class.getName());
// Make sure the props is propagated
@@ -2159,7 +2240,8 @@ protected void setDefaults() {
writeConfig.setDefaultOnCondition(!isCompactionConfigSet,
HoodieCompactionConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
writeConfig.setDefaultOnCondition(!isClusteringConfigSet,
- HoodieClusteringConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
+ HoodieClusteringConfig.newBuilder().withEngineType(engineType)
+ .fromProperties(writeConfig.getProps()).build());
writeConfig.setDefaultOnCondition(!isMetricsConfigSet, HoodieMetricsConfig.newBuilder().fromProperties(
writeConfig.getProps()).build());
writeConfig.setDefaultOnCondition(!isBootstrapConfigSet,
@@ -2175,11 +2257,13 @@ protected void setDefaults() {
writeConfig.setDefaultOnCondition(!isPayloadConfigSet,
HoodiePayloadConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
writeConfig.setDefaultOnCondition(!isMetadataConfigSet,
- HoodieMetadataConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
+ HoodieMetadataConfig.newBuilder().withEngineType(engineType).fromProperties(writeConfig.getProps()).build());
writeConfig.setDefaultOnCondition(!isLockConfigSet,
HoodieLockConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
writeConfig.setDefaultOnCondition(!isPreCommitValidationConfigSet,
HoodiePreCommitValidatorConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
+ writeConfig.setDefaultOnCondition(!isLayoutConfigSet,
+ HoodieLayoutConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
writeConfig.setDefaultValue(TIMELINE_LAYOUT_VERSION_NUM, String.valueOf(TimelineLayoutVersion.CURR_VERSION));
}
@@ -2201,5 +2285,18 @@ public HoodieWriteConfig build() {
// Build WriteConfig at the end
return new HoodieWriteConfig(engineType, writeConfig.getProps());
}
+
+ private String getDefaultMarkersType(EngineType engineType) {
+ switch (engineType) {
+ case SPARK:
+ return MarkerType.TIMELINE_SERVER_BASED.toString();
+ case FLINK:
+ case JAVA:
+ // Timeline-server-based marker is not supported for Flink and Java engines
+ return MarkerType.DIRECT.toString();
+ default:
+ throw new HoodieNotSupportedException("Unsupported engine " + engineType);
+ }
+ }
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsConfig.java
index 60369d72fb4f..8396540394b8 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsConfig.java
@@ -22,6 +22,9 @@
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.table.HoodieTableConfig;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieMetricsCloudWatchConfig;
import org.apache.hudi.metrics.MetricsReporterType;
import javax.annotation.concurrent.Immutable;
@@ -62,6 +65,18 @@ public class HoodieMetricsConfig extends HoodieConfig {
.sinceVersion("0.6.0")
.withDocumentation("");
+ public static final ConfigProperty METRICS_REPORTER_PREFIX = ConfigProperty
+ .key(METRIC_PREFIX + ".reporter.metricsname.prefix")
+ .defaultValue("")
+ .sinceVersion("0.11.0")
+ .withInferFunction(cfg -> {
+ if (cfg.contains(HoodieTableConfig.NAME)) {
+ return Option.of(cfg.getString(HoodieTableConfig.NAME));
+ }
+ return Option.empty();
+ })
+ .withDocumentation("The prefix given to the metrics names.");
+
// Enable metrics collection from executors
public static final ConfigProperty EXECUTOR_METRICS_ENABLE = ConfigProperty
.key(METRIC_PREFIX + ".executor.enable")
@@ -165,6 +180,8 @@ public HoodieMetricsConfig build() {
HoodieMetricsJmxConfig.newBuilder().fromProperties(hoodieMetricsConfig.getProps()).build());
hoodieMetricsConfig.setDefaultOnCondition(reporterType == MetricsReporterType.GRAPHITE,
HoodieMetricsGraphiteConfig.newBuilder().fromProperties(hoodieMetricsConfig.getProps()).build());
+ hoodieMetricsConfig.setDefaultOnCondition(reporterType == MetricsReporterType.CLOUDWATCH,
+ HoodieMetricsCloudWatchConfig.newBuilder().fromProperties(hoodieMetricsConfig.getProps()).build());
return hoodieMetricsConfig;
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsPrometheusConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsPrometheusConfig.java
index c04e8aa1e980..e27ff1bcb089 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsPrometheusConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsPrometheusConfig.java
@@ -166,11 +166,6 @@ public static HoodieMetricsPrometheusConfig.Builder newBuilder() {
return new HoodieMetricsPrometheusConfig.Builder();
}
- @Override
- public Properties getProps() {
- return super.getProps();
- }
-
public static class Builder {
private HoodieMetricsPrometheusConfig hoodieMetricsPrometheusConfig = new HoodieMetricsPrometheusConfig();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java
index 974f4d546b10..0428d12c4030 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java
@@ -28,6 +28,7 @@
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieIndexException;
import org.apache.hudi.exception.HoodieNotSupportedException;
@@ -122,6 +123,20 @@ public abstract HoodieData updateLocation(
@PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
public abstract boolean isImplicitWithStorage();
+ /**
+ * If the `getCustomizedPartitioner` returns a partitioner, it has to be true.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ public boolean requiresTagging(WriteOperationType operationType) {
+ switch (operationType) {
+ case DELETE:
+ case UPSERT:
+ return true;
+ default:
+ return false;
+ }
+ }
+
/**
* Each index type should implement it's own logic to release any resources acquired during the process.
*/
@@ -129,6 +144,6 @@ public void close() {
}
public enum IndexType {
- HBASE, INMEMORY, BLOOM, GLOBAL_BLOOM, SIMPLE, GLOBAL_SIMPLE
+ HBASE, INMEMORY, BLOOM, GLOBAL_BLOOM, SIMPLE, GLOBAL_SIMPLE, BUCKET
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java
index 50d31f9cc50d..18b094890081 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java
@@ -46,8 +46,8 @@ class IntervalTreeBasedGlobalIndexFileFilter implements IndexFileFilter {
IntervalTreeBasedGlobalIndexFileFilter(final Map> partitionToFileIndexInfo) {
List allIndexFiles = new ArrayList<>();
- partitionToFileIndexInfo.forEach((parition, bloomIndexFileInfoList) -> bloomIndexFileInfoList.forEach(file -> {
- fileIdToPartitionPathMap.put(file.getFileId(), parition);
+ partitionToFileIndexInfo.forEach((partition, bloomIndexFileInfoList) -> bloomIndexFileInfoList.forEach(file -> {
+ fileIdToPartitionPathMap.put(file.getFileId(), partition);
allIndexFiles.add(file);
}));
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
new file mode 100644
index 000000000000..7dee9f3cdfa3
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
@@ -0,0 +1,74 @@
+/*
+ * 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.index.bucket;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+public class BucketIdentifier {
+ // compatible with the spark bucket name
+ private static final Pattern BUCKET_NAME = Pattern.compile(".*_(\\d+)(?:\\..*)?$");
+
+ public static int getBucketId(HoodieRecord record, String indexKeyFields, int numBuckets) {
+ return getBucketId(record.getKey(), indexKeyFields, numBuckets);
+ }
+
+ public static int getBucketId(HoodieKey hoodieKey, String indexKeyFields, int numBuckets) {
+ List hashKeyFields;
+ if (!hoodieKey.getRecordKey().contains(":")) {
+ hashKeyFields = Arrays.asList(hoodieKey.getRecordKey());
+ } else {
+ Map recordKeyPairs = Arrays.stream(hoodieKey.getRecordKey().split(","))
+ .map(p -> p.split(":"))
+ .collect(Collectors.toMap(p -> p[0], p -> p[1]));
+ hashKeyFields = Arrays.stream(indexKeyFields.split(","))
+ .map(f -> recordKeyPairs.get(f))
+ .collect(Collectors.toList());
+ }
+ return (hashKeyFields.hashCode() & Integer.MAX_VALUE) % numBuckets;
+ }
+
+ // only for test
+ public static int getBucketId(List hashKeyFields, int numBuckets) {
+ return hashKeyFields.hashCode() % numBuckets;
+ }
+
+ public static int bucketIdFromFileId(String fileId) {
+ return Integer.parseInt(fileId.substring(0, 8));
+ }
+
+ public static String bucketIdStr(int n) {
+ return String.format("%08d", n);
+ }
+
+ public static String newBucketFileIdPrefix(String bucketId) {
+ return FSUtils.createNewFileIdPfx().replaceFirst(".{8}", bucketId);
+ }
+
+ public static boolean isBucketFileName(String name) {
+ return BUCKET_NAME.matcher(name).matches();
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java
new file mode 100644
index 000000000000..acb06ea48bed
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java
@@ -0,0 +1,166 @@
+/*
+ * 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.index.bucket;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.utils.LazyIterableIterator;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Hash indexing mechanism.
+ * @param
+ */
+public class HoodieBucketIndex>
+ extends HoodieIndex {
+
+ private static final Logger LOG = LogManager.getLogger(HoodieBucketIndex.class);
+
+ private final int numBuckets;
+
+ public HoodieBucketIndex(HoodieWriteConfig config) {
+ super(config);
+ numBuckets = config.getBucketIndexNumBuckets();
+ LOG.info("use bucket index, numBuckets=" + numBuckets);
+ }
+
+ @Override
+ public HoodieData updateLocation(HoodieData writeStatuses,
+ HoodieEngineContext context,
+ HoodieTable hoodieTable)
+ throws HoodieIndexException {
+ return writeStatuses;
+ }
+
+ @Override
+ public HoodieData> tagLocation(HoodieData> records,
+ HoodieEngineContext context,
+ HoodieTable hoodieTable)
+ throws HoodieIndexException {
+ HoodieData> taggedRecords = records.mapPartitions(recordIter -> {
+ // partitionPath -> bucketId -> fileInfo
+ Map>> partitionPathFileIDList = new HashMap<>();
+ return new LazyIterableIterator, HoodieRecord>(recordIter) {
+
+ @Override
+ protected void start() {
+
+ }
+
+ @Override
+ protected HoodieRecord computeNext() {
+ HoodieRecord record = recordIter.next();
+ int bucketId = BucketIdentifier.getBucketId(record, config.getBucketIndexHashField(), numBuckets);
+ String partitionPath = record.getPartitionPath();
+ if (!partitionPathFileIDList.containsKey(partitionPath)) {
+ partitionPathFileIDList.put(partitionPath, loadPartitionBucketIdFileIdMapping(hoodieTable, partitionPath));
+ }
+ if (partitionPathFileIDList.get(partitionPath).containsKey(bucketId)) {
+ Pair fileInfo = partitionPathFileIDList.get(partitionPath).get(bucketId);
+ return HoodieIndexUtils.getTaggedRecord(record, Option.of(
+ new HoodieRecordLocation(fileInfo.getRight(), fileInfo.getLeft())
+ ));
+ }
+ return record;
+ }
+
+ @Override
+ protected void end() {
+
+ }
+ };
+ }, true);
+ return taggedRecords;
+ }
+
+ private Map> loadPartitionBucketIdFileIdMapping(
+ HoodieTable hoodieTable,
+ String partition) {
+ // bucketId -> fileIds
+ Map> fileIDList = new HashMap<>();
+ HoodieIndexUtils
+ .getLatestBaseFilesForPartition(partition, hoodieTable)
+ .forEach(file -> {
+ String fileId = file.getFileId();
+ String commitTime = file.getCommitTime();
+ int bucketId = BucketIdentifier.bucketIdFromFileId(fileId);
+ if (!fileIDList.containsKey(bucketId)) {
+ fileIDList.put(bucketId, Pair.of(fileId, commitTime));
+ } else {
+ // check if bucket data is valid
+ throw new HoodieIOException("Find multiple files at partition path="
+ + partition + " belongs to the same bucket id = " + bucketId);
+ }
+ });
+ return fileIDList;
+ }
+
+ @Override
+ public boolean rollbackCommit(String instantTime) {
+ return true;
+ }
+
+ @Override
+ public boolean isGlobal() {
+ return false;
+ }
+
+ @Override
+ public boolean canIndexLogFiles() {
+ return false;
+ }
+
+ @Override
+ public boolean isImplicitWithStorage() {
+ return true;
+ }
+
+ @Override
+ public boolean requiresTagging(WriteOperationType operationType) {
+ switch (operationType) {
+ case INSERT:
+ case INSERT_OVERWRITE:
+ case UPSERT:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ public int getNumBuckets() {
+ return numBuckets;
+ }
+}
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 a33383a05c02..7cc0c5dfa465 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
@@ -29,10 +29,10 @@
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieOperation;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
+import org.apache.hudi.common.model.HoodiePayloadProps;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.model.HoodieRecordPayload;
-import org.apache.hudi.common.model.HoodiePayloadProps;
import org.apache.hudi.common.model.HoodieWriteStat.RuntimeStats;
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.log.AppendResult;
@@ -178,7 +178,7 @@ private void init(HoodieRecord record) {
LOG.error("Error in update task at commit " + instantTime, e);
writeStatus.setGlobalError(e);
throw new HoodieUpsertException("Failed to initialize HoodieAppendHandle for FileId: " + fileId + " on commit "
- + instantTime + " on HDFS path " + hoodieTable.getMetaClient().getBasePath() + partitionPath, e);
+ + instantTime + " on HDFS path " + hoodieTable.getMetaClient().getBasePath() + "/" + partitionPath, e);
}
doInit = false;
}
@@ -360,7 +360,12 @@ protected void appendDataAndDeleteBlocks(Map header)
header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, writeSchemaWithMetaFields.toString());
List blocks = new ArrayList<>(2);
if (recordList.size() > 0) {
- blocks.add(HoodieDataBlock.getBlock(hoodieTable.getLogDataBlockFormat(), recordList, header));
+ if (config.populateMetaFields()) {
+ blocks.add(HoodieDataBlock.getBlock(hoodieTable.getLogDataBlockFormat(), recordList, header));
+ } else {
+ final String keyField = hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();
+ blocks.add(HoodieDataBlock.getBlock(hoodieTable.getLogDataBlockFormat(), recordList, header, keyField));
+ }
}
if (keysToDelete.size() > 0) {
blocks.add(new HoodieDeleteBlock(keysToDelete.toArray(new HoodieKey[keysToDelete.size()]), header));
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 01ad45342c50..22fa31539d88 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
@@ -119,7 +119,8 @@ public HoodieCreateHandle(HoodieWriteConfig config, String instantTime, HoodieTa
@Override
public boolean canWrite(HoodieRecord record) {
- return fileWriter.canWrite() && record.getPartitionPath().equals(writeStatus.getPartitionPath());
+ return (fileWriter.canWrite() && record.getPartitionPath().equals(writeStatus.getPartitionPath()))
+ || layoutControlsNumFiles();
}
/**
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 b13b561f5dff..d1d67efff4b9 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
@@ -250,11 +250,17 @@ protected void init(String fileId, Iterator> newRecordsItr) {
+ ((ExternalSpillableMap) keyToNewRecords).getSizeOfFileOnDiskInBytes());
}
- private boolean writeUpdateRecord(HoodieRecord hoodieRecord, Option indexedRecord) {
+ private boolean writeUpdateRecord(HoodieRecord hoodieRecord, GenericRecord oldRecord, Option indexedRecord) {
+ boolean isDelete = false;
if (indexedRecord.isPresent()) {
updatedRecordsWritten++;
+ GenericRecord record = (GenericRecord) indexedRecord.get();
+ if (oldRecord != record) {
+ // the incoming record is chosen
+ isDelete = HoodieOperation.isDelete(hoodieRecord.getOperation());
+ }
}
- return writeRecord(hoodieRecord, indexedRecord);
+ return writeRecord(hoodieRecord, indexedRecord, isDelete);
}
protected void writeInsertRecord(HoodieRecord hoodieRecord) throws IOException {
@@ -264,12 +270,16 @@ protected void writeInsertRecord(HoodieRecord hoodieRecord) throws IOExceptio
if (insertRecord.isPresent() && insertRecord.get().equals(IGNORE_RECORD)) {
return;
}
- if (writeRecord(hoodieRecord, insertRecord)) {
+ if (writeRecord(hoodieRecord, insertRecord, HoodieOperation.isDelete(hoodieRecord.getOperation()))) {
insertRecordsWritten++;
}
}
protected boolean writeRecord(HoodieRecord hoodieRecord, Option indexedRecord) {
+ return writeRecord(hoodieRecord, indexedRecord, false);
+ }
+
+ protected boolean writeRecord(HoodieRecord hoodieRecord, Option indexedRecord, boolean isDelete) {
Option recordMetadata = hoodieRecord.getData().getMetadata();
if (!partitionPath.equals(hoodieRecord.getPartitionPath())) {
HoodieUpsertException failureEx = new HoodieUpsertException("mismatched partition path, record partition: "
@@ -277,11 +287,8 @@ protected boolean writeRecord(HoodieRecord hoodieRecord, Option HoodieFi
BloomFilter filter = createBloomFilter(config);
HoodieHFileConfig hfileConfig = new HoodieHFileConfig(hoodieTable.getHadoopConf(),
- config.getHFileCompressionAlgorithm(), config.getHFileBlockSize(), config.getHFileMaxFileSize(), filter);
+ config.getHFileCompressionAlgorithm(), config.getHFileBlockSize(), config.getHFileMaxFileSize(),
+ PREFETCH_ON_OPEN, CACHE_DATA_IN_L1, DROP_BEHIND_CACHE_COMPACTION, filter, HFILE_COMPARATOR);
- return new HoodieHFileWriter<>(instantTime, path, hfileConfig, schema, taskContextSupplier);
+ return new HoodieHFileWriter<>(instantTime, path, hfileConfig, schema, taskContextSupplier, config.populateMetaFields());
}
private static HoodieFileWriter newOrcFileWriter(
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileConfig.java
index 031f92cccdfa..7e4c519a8faf 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileConfig.java
@@ -18,35 +18,35 @@
package org.apache.hudi.io.storage;
+import org.apache.hudi.common.bloom.BloomFilter;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hudi.common.bloom.BloomFilter;
public class HoodieHFileConfig {
- private Compression.Algorithm compressionAlgorithm;
- private int blockSize;
- private long maxFileSize;
- private boolean prefetchBlocksOnOpen;
- private boolean cacheDataInL1;
- private boolean dropBehindCacheCompaction;
- private Configuration hadoopConf;
- private BloomFilter bloomFilter;
-
+ public static final KeyValue.KVComparator HFILE_COMPARATOR = new HoodieHBaseKVComparator();
+ public static final boolean PREFETCH_ON_OPEN = CacheConfig.DEFAULT_PREFETCH_ON_OPEN;
+ public static final boolean CACHE_DATA_IN_L1 = HColumnDescriptor.DEFAULT_CACHE_DATA_IN_L1;
// This is private in CacheConfig so have been copied here.
- private static boolean DROP_BEHIND_CACHE_COMPACTION_DEFAULT = true;
-
- public HoodieHFileConfig(Configuration hadoopConf, Compression.Algorithm compressionAlgorithm, int blockSize,
- long maxFileSize, BloomFilter bloomFilter) {
- this(hadoopConf, compressionAlgorithm, blockSize, maxFileSize, CacheConfig.DEFAULT_PREFETCH_ON_OPEN,
- HColumnDescriptor.DEFAULT_CACHE_DATA_IN_L1, DROP_BEHIND_CACHE_COMPACTION_DEFAULT, bloomFilter);
- }
+ public static final boolean DROP_BEHIND_CACHE_COMPACTION = true;
+
+ private final Compression.Algorithm compressionAlgorithm;
+ private final int blockSize;
+ private final long maxFileSize;
+ private final boolean prefetchBlocksOnOpen;
+ private final boolean cacheDataInL1;
+ private final boolean dropBehindCacheCompaction;
+ private final Configuration hadoopConf;
+ private final BloomFilter bloomFilter;
+ private final KeyValue.KVComparator hfileComparator;
public HoodieHFileConfig(Configuration hadoopConf, Compression.Algorithm compressionAlgorithm, int blockSize,
long maxFileSize, boolean prefetchBlocksOnOpen, boolean cacheDataInL1,
- boolean dropBehindCacheCompaction, BloomFilter bloomFilter) {
+ boolean dropBehindCacheCompaction, BloomFilter bloomFilter, KeyValue.KVComparator hfileComparator) {
this.hadoopConf = hadoopConf;
this.compressionAlgorithm = compressionAlgorithm;
this.blockSize = blockSize;
@@ -55,6 +55,7 @@ public HoodieHFileConfig(Configuration hadoopConf, Compression.Algorithm compres
this.cacheDataInL1 = cacheDataInL1;
this.dropBehindCacheCompaction = dropBehindCacheCompaction;
this.bloomFilter = bloomFilter;
+ this.hfileComparator = hfileComparator;
}
public Configuration getHadoopConf() {
@@ -92,4 +93,8 @@ public boolean useBloomFilter() {
public BloomFilter getBloomFilter() {
return bloomFilter;
}
+
+ public KeyValue.KVComparator getHfileComparator() {
+ return hfileComparator;
+ }
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileWriter.java
index 6747c4a4a80e..a719bcb8f334 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileWriter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileWriter.java
@@ -25,6 +25,7 @@
import org.apache.hudi.common.fs.HoodieWrapperFileSystem;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
+
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
@@ -61,6 +62,7 @@ public class HoodieHFileWriter boolean isBootstrapNeeded(Option rollbackedTimestamps = Collections.emptyList();
- if (actionMetadata.isPresent() && actionMetadata.get() instanceof HoodieRollbackMetadata) {
- isRollbackAction = true;
- List rollbackedInstants =
- ((HoodieRollbackMetadata) actionMetadata.get()).getInstantsRollback();
- rollbackedTimestamps = rollbackedInstants.stream().map(instant -> {
- return instant.getCommitTime().toString();
- }).collect(Collectors.toList());
- }
-
+ // Detect the commit gaps if any from the data and the metadata active timeline
if (dataMetaClient.getActiveTimeline().getAllCommitsTimeline().isBeforeTimelineStarts(
latestMetadataInstant.get().getTimestamp())
- && (!isRollbackAction || !rollbackedTimestamps.contains(latestMetadataInstantTimestamp))) {
- LOG.warn("Metadata Table will need to be re-bootstrapped as un-synced instants have been archived."
+ && !isCommitRevertedByInFlightAction(actionMetadata, latestMetadataInstantTimestamp)) {
+ LOG.error("Metadata Table will need to be re-bootstrapped as un-synced instants have been archived."
+ " latestMetadataInstant=" + latestMetadataInstant.get().getTimestamp()
+ ", latestDataInstant=" + dataMetaClient.getActiveTimeline().firstInstant().get().getTimestamp());
return true;
@@ -359,10 +370,59 @@ private boolean isBootstrapNeeded(Option - ActionMetadata type
+ * @return True if the latest instant action is reverted by the action
+ */
+ private boolean isCommitRevertedByInFlightAction(Option actionMetadata,
+ final String latestMetadataInstantTimestamp) {
+ if (!actionMetadata.isPresent()) {
+ return false;
+ }
+
+ final String INSTANT_ACTION = (actionMetadata.get() instanceof HoodieRollbackMetadata
+ ? HoodieTimeline.ROLLBACK_ACTION
+ : (actionMetadata.get() instanceof HoodieRestoreMetadata ? HoodieTimeline.RESTORE_ACTION : ""));
+
+ List affectedInstantTimestamps;
+ switch (INSTANT_ACTION) {
+ case HoodieTimeline.ROLLBACK_ACTION:
+ List rollbackedInstants =
+ ((HoodieRollbackMetadata) actionMetadata.get()).getInstantsRollback();
+ affectedInstantTimestamps = rollbackedInstants.stream().map(instant -> {
+ return instant.getCommitTime().toString();
+ }).collect(Collectors.toList());
+
+ if (affectedInstantTimestamps.contains(latestMetadataInstantTimestamp)) {
+ return true;
+ }
+ break;
+ case HoodieTimeline.RESTORE_ACTION:
+ List restoredInstants =
+ ((HoodieRestoreMetadata) actionMetadata.get()).getRestoreInstantInfo();
+ affectedInstantTimestamps = restoredInstants.stream().map(instant -> {
+ return instant.getCommitTime().toString();
+ }).collect(Collectors.toList());
+
+ if (affectedInstantTimestamps.contains(latestMetadataInstantTimestamp)) {
+ return true;
+ }
+ break;
+ default:
+ return false;
+ }
+
+ return false;
+ }
+
/**
* Initialize the Metadata Table by listing files and partitions from the file system.
*
- * @param dataMetaClient {@code HoodieTableMetaClient} for the dataset.
+ * @param dataMetaClient {@code HoodieTableMetaClient} for the dataset.
* @param inflightInstantTimestamp
*/
private boolean bootstrapFromFilesystem(HoodieEngineContext engineContext, HoodieTableMetaClient dataMetaClient,
@@ -392,15 +452,19 @@ private boolean bootstrapFromFilesystem(HoodieEngineContext engineContext, Hoodi
.setTableType(HoodieTableType.MERGE_ON_READ)
.setTableName(tableName)
.setArchiveLogFolder(ARCHIVELOG_FOLDER.defaultValue())
- .setPayloadClassName(HoodieMetadataPayload.class.getName())
- .setBaseFileFormat(HoodieFileFormat.HFILE.toString())
- .initTable(hadoopConf.get(), metadataWriteConfig.getBasePath());
+ .setPayloadClassName(HoodieMetadataPayload.class.getName())
+ .setBaseFileFormat(HoodieFileFormat.HFILE.toString())
+ .setRecordKeyFields(RECORD_KEY_FIELD)
+ .setPopulateMetaFields(dataWriteConfig.getMetadataConfig().populateMetaFields())
+ .setKeyGeneratorClassProp(HoodieTableMetadataKeyGenerator.class.getCanonicalName())
+ .initTable(hadoopConf.get(), metadataWriteConfig.getBasePath());
initTableMetadata();
initializeFileGroups(dataMetaClient, MetadataPartitionType.FILES, createInstantTime, 1);
// List all partitions in the basePath of the containing dataset
LOG.info("Initializing metadata table by using file listings in " + dataWriteConfig.getBasePath());
+ engineContext.setJobStatus(this.getClass().getSimpleName(), "Bootstrap: initializing metadata table by listing files and partitions");
List dirInfoList = listAllPartitions(dataMetaClient);
// During bootstrap, the list of files to be committed can be huge. So creating a HoodieCommitMetadata out of these
@@ -619,7 +683,10 @@ public void close() throws Exception {
* deltacommit.
*/
protected void compactIfNecessary(AbstractHoodieWriteClient writeClient, String instantTime) {
- String latestDeltacommitTime = metadataMetaClient.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().lastInstant()
+ // finish off any pending compactions if any from previous attempt.
+ writeClient.runAnyPendingCompactions();
+
+ String latestDeltacommitTime = metadataMetaClient.reloadActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().lastInstant()
.get().getTimestamp();
List pendingInstants = dataMetaClient.reloadActiveTimeline().filterInflightsAndRequested()
.findInstantsBefore(latestDeltacommitTime).getInstants().collect(Collectors.toList());
@@ -639,7 +706,20 @@ protected void compactIfNecessary(AbstractHoodieWriteClient writeClient, String
}
}
- protected void doClean(AbstractHoodieWriteClient writeClient, String instantTime) {
+ protected void cleanIfNecessary(AbstractHoodieWriteClient writeClient, String instantTime) {
+ Option lastCompletedCompactionInstant = metadataMetaClient.reloadActiveTimeline()
+ .getCommitTimeline().filterCompletedInstants().lastInstant();
+ if (lastCompletedCompactionInstant.isPresent()
+ && metadataMetaClient.getActiveTimeline().filterCompletedInstants()
+ .findInstantsAfter(lastCompletedCompactionInstant.get().getTimestamp()).countInstants() < 3) {
+ // do not clean the log files immediately after compaction to give some buffer time for metadata table reader,
+ // because there is case that the reader has prepared for the log file readers already before the compaction completes
+ // while before/during the reading of the log files, the cleaning triggers and delete the reading files,
+ // then a FileNotFoundException(for LogFormatReader) or NPE(for HFileReader) would throw.
+
+ // 3 is a value that I think is enough for metadata table reader.
+ return;
+ }
// Trigger cleaning with suffixes based on the same instant time. This ensures that any future
// delta commits synced over will not have an instant time lesser than the last completed instant on the
// metadata table.
@@ -652,7 +732,8 @@ protected void doClean(AbstractHoodieWriteClient writeClient, String instantTime
*
*/
protected void bootstrapCommit(List partitionInfoList, String createInstantTime) {
- List partitions = partitionInfoList.stream().map(p -> p.getRelativePath()).collect(Collectors.toList());
+ List partitions = partitionInfoList.stream().map(p ->
+ p.getRelativePath().isEmpty() ? NON_PARTITIONED_NAME : p.getRelativePath()).collect(Collectors.toList());
final int totalFiles = partitionInfoList.stream().mapToInt(p -> p.getTotalFiles()).sum();
// Record which saves the list of all partitions
@@ -667,7 +748,7 @@ protected void bootstrapCommit(List partitionInfoList, String cre
HoodieData fileListRecords = engineContext.parallelize(partitionInfoList, partitionInfoList.size()).map(partitionInfo -> {
// Record which saves files within a partition
return HoodieMetadataPayload.createPartitionFilesRecord(
- partitionInfo.getRelativePath(), Option.of(partitionInfo.getFileNameToSizeMap()), Option.empty());
+ partitionInfo.getRelativePath().isEmpty() ? NON_PARTITIONED_NAME : partitionInfo.getRelativePath(), Option.of(partitionInfo.getFileNameToSizeMap()), Option.empty());
});
partitionRecords = partitionRecords.union(fileListRecords);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataKeyGenerator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataKeyGenerator.java
new file mode 100644
index 000000000000..e9d7aec8c8e2
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataKeyGenerator.java
@@ -0,0 +1,52 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.keygen.KeyGenUtils;
+
+/**
+ * Custom key generator for the Hoodie table metadata. The metadata table record payload
+ * has an internal schema with a known key field HoodieMetadataPayload.SCHEMA_FIELD_ID_KEY.
+ * With or without the virtual keys, getting the key from the metadata table record is always
+ * via the above field and there is no real need for a key generator. But, when a write
+ * client is instantiated for the metadata table, when virtual keys are enabled, and when
+ * key generator class is not configured, the default SimpleKeyGenerator will be used.
+ * To avoid using any other key generators for the metadata table which rely on certain
+ * config properties, we need this custom key generator exclusively for the metadata table.
+ */
+public class HoodieTableMetadataKeyGenerator extends BaseKeyGenerator {
+
+ public HoodieTableMetadataKeyGenerator(TypedProperties config) {
+ super(config);
+ }
+
+ @Override
+ public String getRecordKey(GenericRecord record) {
+ return KeyGenUtils.getRecordKey(record, HoodieMetadataPayload.SCHEMA_FIELD_ID_KEY);
+ }
+
+ @Override
+ public String getPartitionPath(GenericRecord record) {
+ return "";
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java
index 67befea42456..e874047b8c64 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java
@@ -215,7 +215,7 @@ public void updateIndexMetrics(final String action, final long durationInMs) {
}
String getMetricsName(String action, String metric) {
- return config == null ? null : String.format("%s.%s.%s", tableName, action, metric);
+ return config == null ? null : String.format("%s.%s.%s", config.getMetricReporterMetricsNamePrefix(), action, metric);
}
/**
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/Metrics.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/Metrics.java
index 1e4d83ab57ee..b570f512f371 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/Metrics.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/Metrics.java
@@ -46,7 +46,7 @@ public class Metrics {
private Metrics(HoodieWriteConfig metricConfig) {
registry = new MetricRegistry();
- commonMetricPrefix = metricConfig.getTableName();
+ commonMetricPrefix = metricConfig.getMetricReporterMetricsNamePrefix();
reporter = MetricsReporterFactory.createReporter(metricConfig, registry);
if (reporter == null) {
throw new RuntimeException("Cannot initialize Reporter.");
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterFactory.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterFactory.java
index 820c1a3e8887..dc9e80431b8d 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterFactory.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterFactory.java
@@ -22,6 +22,7 @@
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metrics.cloudwatch.CloudWatchMetricsReporter;
import org.apache.hudi.metrics.datadog.DatadogMetricsReporter;
import com.codahale.metrics.MetricRegistry;
@@ -77,6 +78,9 @@ public static MetricsReporter createReporter(HoodieWriteConfig config, MetricReg
case CONSOLE:
reporter = new ConsoleMetricsReporter(registry);
break;
+ case CLOUDWATCH:
+ reporter = new CloudWatchMetricsReporter(config, registry);
+ break;
default:
LOG.error("Reporter type[" + type + "] is not supported.");
break;
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterType.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterType.java
index 36b15a89ac88..3c8600159287 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterType.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/MetricsReporterType.java
@@ -22,5 +22,5 @@
* Types of the reporter supported, hudi also supports user defined reporter.
*/
public enum MetricsReporterType {
- GRAPHITE, INMEMORY, JMX, DATADOG, CONSOLE, PROMETHEUS_PUSHGATEWAY, PROMETHEUS
+ GRAPHITE, INMEMORY, JMX, DATADOG, CONSOLE, PROMETHEUS_PUSHGATEWAY, PROMETHEUS, CLOUDWATCH
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/cloudwatch/CloudWatchMetricsReporter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/cloudwatch/CloudWatchMetricsReporter.java
new file mode 100644
index 000000000000..c7134a10360f
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/cloudwatch/CloudWatchMetricsReporter.java
@@ -0,0 +1,85 @@
+/*
+ * 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.metrics.cloudwatch;
+
+import org.apache.hudi.aws.cloudwatch.CloudWatchReporter;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.metrics.MetricsReporter;
+
+import com.codahale.metrics.MetricRegistry;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.Closeable;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Hudi Amazon CloudWatch metrics reporter. Responsible for reading Hoodie metrics configurations and hooking up with
+ * {@link org.apache.hudi.metrics.Metrics}. Internally delegates reporting tasks to {@link CloudWatchReporter}.
+ */
+public class CloudWatchMetricsReporter extends MetricsReporter {
+
+ private static final Logger LOG = LogManager.getLogger(CloudWatchMetricsReporter.class);
+
+ private final MetricRegistry registry;
+ private final HoodieWriteConfig config;
+ private final CloudWatchReporter reporter;
+
+ public CloudWatchMetricsReporter(HoodieWriteConfig config, MetricRegistry registry) {
+ this.config = config;
+ this.registry = registry;
+ this.reporter = createCloudWatchReporter();
+ }
+
+ CloudWatchMetricsReporter(HoodieWriteConfig config, MetricRegistry registry, CloudWatchReporter reporter) {
+ this.config = config;
+ this.registry = registry;
+ this.reporter = reporter;
+ }
+
+ private CloudWatchReporter createCloudWatchReporter() {
+ return CloudWatchReporter.forRegistry(registry)
+ .prefixedWith(config.getCloudWatchMetricPrefix())
+ .namespace(config.getCloudWatchMetricNamespace())
+ .maxDatumsPerRequest(config.getCloudWatchMaxDatumsPerRequest())
+ .build(config.getProps());
+ }
+
+ @Override
+ public void start() {
+ LOG.info("Starting CloudWatch Metrics Reporter.");
+ reporter.start(config.getCloudWatchReportPeriodSeconds(), TimeUnit.SECONDS);
+ }
+
+ @Override
+ public void report() {
+ reporter.report();
+ }
+
+ @Override
+ public Closeable getReporter() {
+ return reporter;
+ }
+
+ @Override
+ public void stop() {
+ LOG.info("Stopping CloudWatch Metrics Reporter.");
+ reporter.stop();
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/optimize/HilbertCurveUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/optimize/HilbertCurveUtils.java
new file mode 100644
index 000000000000..0f216abeee74
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/optimize/HilbertCurveUtils.java
@@ -0,0 +1,52 @@
+/*
+ * 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.optimize;
+
+import org.davidmoten.hilbert.HilbertCurve;
+
+import java.math.BigInteger;
+
+/**
+ * Utils for Hilbert Curve.
+ */
+public class HilbertCurveUtils {
+ public static byte[] indexBytes(HilbertCurve hilbertCurve, long[] points, int paddingNum) {
+ BigInteger index = hilbertCurve.index(points);
+ return paddingToNByte(index.toByteArray(), paddingNum);
+ }
+
+ public static byte[] paddingToNByte(byte[] a, int paddingNum) {
+ if (a.length == paddingNum) {
+ return a;
+ }
+ if (a.length > paddingNum) {
+ byte[] result = new byte[paddingNum];
+ System.arraycopy(a, 0, result, 0, paddingNum);
+ return result;
+ }
+ int paddingSize = paddingNum - a.length;
+ byte[] result = new byte[paddingNum];
+ for (int i = 0; i < paddingSize; i++) {
+ result[i] = 0;
+ }
+ System.arraycopy(a, 0, result, paddingSize, a.length);
+ return result;
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/FileIdPrefixProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/FileIdPrefixProvider.java
index d06da9b0d9b9..0fc0823184ea 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/FileIdPrefixProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/FileIdPrefixProvider.java
@@ -18,17 +18,17 @@
package org.apache.hudi.table;
-import java.util.Properties;
+import org.apache.hudi.common.config.TypedProperties;
public abstract class FileIdPrefixProvider {
- private final Properties props;
+ private final TypedProperties props;
- public FileIdPrefixProvider(Properties props) {
+ public FileIdPrefixProvider(TypedProperties props) {
this.props = props;
}
- public Properties getProps() {
+ public TypedProperties getProps() {
return props;
}
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 6046374ba107..40e3a316db10 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
@@ -74,9 +74,12 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hudi.table.storage.HoodieLayoutFactory;
+import org.apache.hudi.table.storage.HoodieStorageLayout;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
+import javax.annotation.Nonnull;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
@@ -105,6 +108,7 @@ public abstract class HoodieTable implem
private SerializableConfiguration hadoopConfiguration;
protected final TaskContextSupplier taskContextSupplier;
private final HoodieTableMetadata metadata;
+ private final HoodieStorageLayout storageLayout;
private transient FileSystemViewManager viewManager;
protected final transient HoodieEngineContext context;
@@ -122,11 +126,16 @@ protected HoodieTable(HoodieWriteConfig config, HoodieEngineContext context, Hoo
this.viewManager = FileSystemViewManager.createViewManager(context, config.getMetadataConfig(), config.getViewStorageConfig(), config.getCommonConfig(), () -> metadata);
this.metaClient = metaClient;
this.index = getIndex(config, context);
+ this.storageLayout = getStorageLayout(config);
this.taskContextSupplier = context.getTaskContextSupplier();
}
protected abstract HoodieIndex getIndex(HoodieWriteConfig config, HoodieEngineContext context);
+ protected HoodieStorageLayout getStorageLayout(HoodieWriteConfig config) {
+ return HoodieLayoutFactory.createLayout(config);
+ }
+
private synchronized FileSystemViewManager getViewManager() {
if (null == viewManager) {
viewManager = FileSystemViewManager.createViewManager(getContext(), config.getMetadataConfig(), config.getViewStorageConfig(), config.getCommonConfig(), () -> metadata);
@@ -245,14 +254,17 @@ public abstract HoodieWriteMetadata bulkInsertPrepped(HoodieEngineContext con
public abstract HoodieWriteMetadata insertOverwriteTable(HoodieEngineContext context, String instantTime, I records);
/**
- * update statistics info for current table.
- * to do adaptation, once RFC-27 is finished.
+ * Updates Metadata Indexes (like Column Stats index)
+ * TODO rebase onto metadata table (post RFC-27)
*
- * @param context HoodieEngineContext
- * @param instantTime Instant time for the replace action
- * @param isOptimizeOperation whether current operation is OPTIMIZE type
+ * @param context instance of {@link HoodieEngineContext}
+ * @param instantTime instant of the carried operation triggering the update
*/
- public abstract void updateStatistics(HoodieEngineContext context, List stats, String instantTime, Boolean isOptimizeOperation);
+ public abstract void updateMetadataIndexes(
+ @Nonnull HoodieEngineContext context,
+ @Nonnull List stats,
+ @Nonnull String instantTime
+ ) throws Exception;
public HoodieWriteConfig getConfig() {
return config;
@@ -361,6 +373,10 @@ public HoodieActiveTimeline getActiveTimeline() {
return index;
}
+ public HoodieStorageLayout getStorageLayout() {
+ return storageLayout;
+ }
+
/**
* Schedule compaction for the instant time.
*
@@ -442,12 +458,13 @@ public abstract Option scheduleCleaning(HoodieEngineContext c
* @param context HoodieEngineContext
* @param instantTime Instant Time for scheduling rollback
* @param instantToRollback instant to be rolled back
+ * @param shouldRollbackUsingMarkers uses marker based rollback strategy when set to true. uses list based rollback when false.
* @return HoodieRollbackPlan containing info on rollback.
*/
public abstract Option scheduleRollback(HoodieEngineContext context,
String instantTime,
HoodieInstant instantToRollback,
- boolean skipTimelinePublish);
+ boolean skipTimelinePublish, boolean shouldRollbackUsingMarkers);
/**
* Rollback the (inflight/committed) record changes with the given commit time.
@@ -490,7 +507,7 @@ public abstract HoodieRestoreMetadata restore(HoodieEngineContext context,
*/
public void rollbackInflightCompaction(HoodieInstant inflightInstant) {
String commitTime = HoodieActiveTimeline.createNewInstantTime();
- scheduleRollback(context, commitTime, inflightInstant, false);
+ scheduleRollback(context, commitTime, inflightInstant, false, config.shouldRollbackUsingMarkers());
rollback(context, commitTime, inflightInstant, false, false);
getActiveTimeline().revertCompactionInflightToRequested(inflightInstant);
}
@@ -732,10 +749,11 @@ public HoodieEngineContext getContext() {
/**
* Get Table metadata writer.
*
+ * @param triggeringInstantTimestamp - The instant that is triggering this metadata write
* @return instance of {@link HoodieTableMetadataWriter
*/
- public final Option getMetadataWriter() {
- return getMetadataWriter(Option.empty());
+ public final Option getMetadataWriter(String triggeringInstantTimestamp) {
+ return getMetadataWriter(triggeringInstantTimestamp, Option.empty());
}
/**
@@ -747,10 +765,19 @@ public final Option getMetadataWriter() {
/**
* Get Table metadata writer.
+ *
+ * Note:
+ * Get the metadata writer for the conf. If the metadata table doesn't exist,
+ * this wil trigger the creation of the table and the initial bootstrapping.
+ * Since this call is under the transaction lock, other concurrent writers
+ * are blocked from doing the similar initial metadata table creation and
+ * the bootstrapping.
*
+ * @param triggeringInstantTimestamp - The instant that is triggering this metadata write
* @return instance of {@link HoodieTableMetadataWriter}
*/
- public Option getMetadataWriter(Option actionMetadata) {
+ public Option getMetadataWriter(String triggeringInstantTimestamp,
+ Option actionMetadata) {
// Each engine is expected to override this and
// provide the actual metadata writer, if enabled.
return Option.empty();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java
index d492fb6577a9..58e03f5f39fc 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java
@@ -18,9 +18,11 @@
package org.apache.hudi.table;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hudi.avro.model.HoodieArchivedMetaEntry;
import org.apache.hudi.client.utils.MetadataConversionUtils;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieArchivedLogFile;
import org.apache.hudi.common.model.HoodieAvroPayload;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
@@ -127,7 +129,7 @@ public boolean archiveIfRequired(HoodieEngineContext context) throws IOException
LOG.info("Archiving instants " + instantsToArchive);
archive(context, instantsToArchive);
LOG.info("Deleting archived instants " + instantsToArchive);
- success = deleteArchivedInstants(instantsToArchive);
+ success = deleteArchivedInstants(instantsToArchive, context);
} else {
LOG.info("No Instants to archive");
}
@@ -154,10 +156,14 @@ private Stream getCleanInstantsToArchive() {
private Stream getCommitInstantsToArchive() {
// TODO (na) : Add a way to return actions associated with a timeline and then merge/unify
- // with logic above to avoid Stream.concats
+ // with logic above to avoid Stream.concat
HoodieTimeline commitTimeline = table.getCompletedCommitsTimeline();
- Option oldestPendingCompactionInstant =
- table.getActiveTimeline().filterPendingCompactionTimeline().firstInstant();
+
+ Option oldestPendingCompactionAndReplaceInstant = table.getActiveTimeline()
+ .getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION))
+ .filter(s -> !s.isCompleted())
+ .firstInstant();
+
Option oldestInflightCommitInstant =
table.getActiveTimeline()
.getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.COMMIT_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION))
@@ -170,11 +176,11 @@ private Stream getCommitInstantsToArchive() {
// Actually do the commits
Stream instantToArchiveStream = commitTimeline.getInstants()
.filter(s -> {
- // if no savepoint present, then dont filter
+ // if no savepoint present, then don't filter
return !(firstSavepoint.isPresent() && HoodieTimeline.compareTimestamps(firstSavepoint.get().getTimestamp(), LESSER_THAN_OR_EQUALS, s.getTimestamp()));
}).filter(s -> {
// Ensure commits >= oldest pending compaction commit is retained
- return oldestPendingCompactionInstant
+ return oldestPendingCompactionAndReplaceInstant
.map(instant -> HoodieTimeline.compareTimestamps(instant.getTimestamp(), GREATER_THAN, s.getTimestamp()))
.orElse(true);
});
@@ -224,19 +230,34 @@ private Stream getInstantsToArchive() {
HoodieInstant.getComparableAction(hoodieInstant.getAction()))).stream());
}
- private boolean deleteArchivedInstants(List archivedInstants) throws IOException {
+ private boolean deleteArchivedInstants(List archivedInstants, HoodieEngineContext context) throws IOException {
LOG.info("Deleting instants " + archivedInstants);
boolean success = true;
- for (HoodieInstant archivedInstant : archivedInstants) {
- Path commitFile = new Path(metaClient.getMetaPath(), archivedInstant.getFileName());
- try {
- if (metaClient.getFs().exists(commitFile)) {
- success &= metaClient.getFs().delete(commitFile, false);
- LOG.info("Archived and deleted instant file " + commitFile);
- }
- } catch (IOException e) {
- throw new HoodieIOException("Failed to delete archived instant " + archivedInstant, e);
- }
+ List instantFiles = archivedInstants.stream().map(archivedInstant ->
+ new Path(metaClient.getMetaPath(), archivedInstant.getFileName())
+ ).map(Path::toString).collect(Collectors.toList());
+
+ context.setJobStatus(this.getClass().getSimpleName(), "Delete archived instants");
+ Map resultDeleteInstantFiles = FSUtils.parallelizeFilesProcess(context,
+ metaClient.getFs(),
+ config.getArchiveDeleteParallelism(),
+ pairOfSubPathAndConf -> {
+ Path commitFile = new Path(pairOfSubPathAndConf.getKey());
+ try {
+ FileSystem fs = commitFile.getFileSystem(pairOfSubPathAndConf.getValue().get());
+ if (fs.exists(commitFile)) {
+ return fs.delete(commitFile, false);
+ }
+ return true;
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to delete archived instant " + commitFile, e);
+ }
+ },
+ instantFiles);
+
+ for (Map.Entry result : resultDeleteInstantFiles.entrySet()) {
+ LOG.info("Archived and deleted instant file " + result.getKey() + " : " + result.getValue());
+ success &= result.getValue();
}
// Remove older meta-data from auxiliary path too
@@ -244,7 +265,7 @@ private boolean deleteArchivedInstants(List