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 0000000000000..426d1cfaf4020
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/InProcessLockProvider.java
@@ -0,0 +1,118 @@
+/*
+ * 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.io.Serializable;
+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, Serializable {
+
+ private static final Logger LOG = LogManager.getLogger(InProcessLockProvider.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 33519b9adc0de..773685980af43 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,14 +20,12 @@
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.HoodieLockConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieLockException;
import org.apache.log4j.LogManager;
@@ -45,15 +43,18 @@ public class LockManager implements Serializable, AutoCloseable {
private final HoodieWriteConfig writeConfig;
private final LockConfiguration lockConfiguration;
private final SerializableConfiguration hadoopConf;
+ private final int maxRetries;
+ private final long maxWaitTimeInMs;
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());
+ maxRetries = lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY,
+ Integer.parseInt(HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.defaultValue()));
+ maxWaitTimeInMs = lockConfiguration.getConfig().getLong(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY,
+ Long.parseLong(HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.defaultValue()));
}
public void lock() {
@@ -61,19 +62,17 @@ public void lock() {
LockProvider lockProvider = getLockProvider();
int retryCount = 0;
boolean acquired = false;
- int retries = lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY);
- long waitTimeInMs = lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY);
- while (retryCount <= retries) {
+ while (retryCount <= maxRetries) {
try {
acquired = lockProvider.tryLock(writeConfig.getLockAcquireWaitTimeoutInMs(), TimeUnit.MILLISECONDS);
if (acquired) {
break;
}
LOG.info("Retrying to acquire lock...");
- Thread.sleep(waitTimeInMs);
+ Thread.sleep(maxWaitTimeInMs);
retryCount++;
- } catch (InterruptedException e) {
- if (retryCount >= retries) {
+ } catch (HoodieLockException | InterruptedException e) {
+ if (retryCount >= maxRetries) {
throw new HoodieLockException("Unable to acquire lock, lock object ", e);
}
}
@@ -100,22 +99,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/MetadataConversionUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/MetadataConversionUtils.java
index 02d2b04ff0caa..8a9d0b3204e74 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/MetadataConversionUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/MetadataConversionUtils.java
@@ -134,7 +134,7 @@ public static Option getInflightReplaceMetadata(HoodieTabl
return Option.of(HoodieCommitMetadata.fromBytes(inflightContent.get(), HoodieCommitMetadata.class));
}
- public static Option getRequestedReplaceMetadata(HoodieTableMetaClient metaClient, HoodieInstant instant) throws IOException {
+ private static Option getRequestedReplaceMetadata(HoodieTableMetaClient metaClient, HoodieInstant instant) throws IOException {
Option requestedContent = metaClient.getActiveTimeline().getInstantDetails(instant);
if (!requestedContent.isPresent() || requestedContent.get().length == 0) {
// requested commit files can be empty in some certain cases, e.g. insert_overwrite or insert_overwrite_table.
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 8e59478da05f9..ed2ea457764fb 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 73c66d8038e5d..057b4a6f61299 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,15 +67,35 @@ 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))
.sinceVersion("0.7.0")
.withDocumentation("Files smaller than the size specified here are candidates for clustering");
+ public static final ConfigProperty PARTITION_REGEX_PATTERN = ConfigProperty
+ .key(CLUSTERING_STRATEGY_PARAM_PREFIX + "partition.regex.pattern")
+ .noDefaultValue()
+ .sinceVersion("0.11.0")
+ .withDocumentation("Filter clustering partitions that matched regex pattern");
+
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 +103,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 +133,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 +189,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 +249,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 +371,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 +396,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 +415,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;
@@ -339,11 +430,26 @@ public Builder withClusteringTargetPartitions(int clusteringTargetPartitions) {
return this;
}
+ public Builder withClusteringPartitionRegexPattern(String pattern) {
+ clusteringConfig.setValue(PARTITION_REGEX_PATTERN, pattern);
+ return this;
+ }
+
public Builder withClusteringSkipPartitionsFromLatest(int clusteringSkipPartitionsFromLatest) {
clusteringConfig.setValue(PLAN_STRATEGY_SKIP_PARTITIONS_FROM_LATEST, String.valueOf(clusteringSkipPartitionsFromLatest));
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 +491,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 +511,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 +542,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 +592,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 ce74aad6b0989..4d1e197cf8a1e 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())
@@ -206,6 +224,12 @@ public class HoodieCompactionConfig extends HoodieConfig {
.withDocumentation("Used by org.apache.hudi.io.compact.strategy.DayBasedCompactionStrategy to denote the number of "
+ "latest partitions to compact during a compaction run.");
+ public static final ConfigProperty PRESERVE_COMMIT_METADATA = ConfigProperty
+ .key("hoodie.compaction.preserve.commit.metadata")
+ .defaultValue(false)
+ .sinceVersion("0.11.0")
+ .withDocumentation("When rewriting data, preserves existing hoodie_commit_time");
+
/**
* Configs related to specific table types.
*/
@@ -231,6 +255,22 @@ public class HoodieCompactionConfig extends HoodieConfig {
+ "record size estimate compute dynamically based on commit metadata. "
+ " This is critical in computing the insert parallelism and bin-packing inserts into small files.");
+ public static final ConfigProperty ARCHIVE_MERGE_FILES_BATCH_SIZE = ConfigProperty
+ .key("hoodie.archive.merge.files.batch.size")
+ .defaultValue(10)
+ .withDocumentation("The number of small archive files to be merged at once.");
+
+ public static final ConfigProperty ARCHIVE_MERGE_SMALL_FILE_LIMIT_BYTES = ConfigProperty
+ .key("hoodie.archive.merge.small.file.limit.bytes")
+ .defaultValue(20L * 1024 * 1024)
+ .withDocumentation("This config sets the archive file size limit below which an archive file becomes a candidate to be selected as such a small file.");
+
+ public static final ConfigProperty ARCHIVE_MERGE_ENABLE = ConfigProperty
+ .key("hoodie.archive.merge.enable")
+ .defaultValue(false)
+ .withDocumentation("When enable, hoodie will auto merge several small archive files into larger one. It's"
+ + " useful when storage scheme doesn't support append operation.");
+
/** @deprecated Use {@link #CLEANER_POLICY} and its methods instead */
@Deprecated
public static final String CLEANER_POLICY_PROP = CLEANER_POLICY.key();
@@ -482,6 +522,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;
@@ -518,6 +563,21 @@ public Builder archiveCommitsWith(int minToKeep, int maxToKeep) {
return this;
}
+ public Builder withArchiveMergeFilesBatchSize(int number) {
+ compactionConfig.setValue(ARCHIVE_MERGE_FILES_BATCH_SIZE, String.valueOf(number));
+ return this;
+ }
+
+ public Builder withArchiveMergeSmallFileLimit(long size) {
+ compactionConfig.setValue(ARCHIVE_MERGE_SMALL_FILE_LIMIT_BYTES, String.valueOf(size));
+ return this;
+ }
+
+ public Builder withArchiveMergeEnable(boolean enable) {
+ compactionConfig.setValue(ARCHIVE_MERGE_ENABLE, String.valueOf(enable));
+ return this;
+ }
+
public Builder compactionSmallFileSize(long smallFileLimitBytes) {
compactionConfig.setValue(PARQUET_SMALL_FILE_LIMIT, String.valueOf(smallFileLimitBytes));
return this;
@@ -568,6 +628,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 +653,16 @@ 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 withPreserveCommitMetadata(boolean preserveCommitMetadata) {
+ compactionConfig.setValue(PRESERVE_COMMIT_METADATA, String.valueOf(preserveCommitMetadata));
+ 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 561460777ff6b..3d7e3a7941daa 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 4f9b9fcd50e84..e156310c736b0 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 0000000000000..1d0810696dc93
--- /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.contains(HoodieIndexConfig.INDEX_TYPE.key()) && 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 fd7b310a57d6e..723c9f3cbff48 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/HoodieStorageConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java
index 22118da471347..42689ec18e948 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java
@@ -116,14 +116,14 @@ public class HoodieStorageConfig extends HoodieConfig {
.withDocumentation("Whether to use dictionary encoding");
public static final ConfigProperty PARQUET_WRITE_LEGACY_FORMAT_ENABLED = ConfigProperty
- .key("hoodie.parquet.writeLegacyFormat.enabled")
+ .key("hoodie.parquet.writelegacyformat.enabled")
.defaultValue("false")
.withDocumentation("Sets spark.sql.parquet.writeLegacyFormat. If true, data will be written in a way of Spark 1.4 and earlier. "
+ "For example, decimal values will be written in Parquet's fixed-length byte array format which other systems such as Apache Hive and Apache Impala use. "
+ "If false, the newer format in Parquet will be used. For example, decimals will be written in int-based format.");
public static final ConfigProperty PARQUET_OUTPUT_TIMESTAMP_TYPE = ConfigProperty
- .key("hoodie.parquet.outputTimestampType")
+ .key("hoodie.parquet.outputtimestamptype")
.defaultValue("TIMESTAMP_MILLIS")
.withDocumentation("Sets spark.sql.parquet.outputTimestampType. Parquet timestamp type to use when Spark writes data to Parquet files.");
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 e167b12c82045..5896ac49ea69d 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
@@ -21,12 +21,14 @@
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.bootstrap.BootstrapMode;
import org.apache.hudi.client.transaction.ConflictResolutionStrategy;
+import org.apache.hudi.client.transaction.lock.InProcessLockProvider;
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.HoodieCommonConfig;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.fs.ConsistencyGuardConfig;
import org.apache.hudi.common.model.HoodieCleaningPolicy;
@@ -45,17 +47,21 @@
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;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.keygen.constant.KeyGeneratorType;
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 +126,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 +254,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")
@@ -279,7 +288,7 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty BULK_INSERT_SORT_MODE = ConfigProperty
.key("hoodie.bulkinsert.sort.mode")
.defaultValue(BulkInsertSortMode.GLOBAL_SORT.toString())
- .withDocumentation("Sorting modes to use for sorting records for bulk insert. This is user when user "
+ .withDocumentation("Sorting modes to use for sorting records for bulk insert. This is use when user "
+ BULKINSERT_USER_DEFINED_PARTITIONER_CLASS_NAME.key() + "is not configured. Available values are - "
+ "GLOBAL_SORT: this ensures best file sizes, with lowest memory overhead at cost of sorting. "
+ "PARTITION_SORT: Strikes a balance by only sorting within a partition, still keeping the memory overhead of writing "
@@ -354,6 +363,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 +387,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.
*/
@@ -885,6 +892,10 @@ public String getKeyGeneratorClass() {
return getString(KEYGENERATOR_CLASS_NAME);
}
+ public boolean isConsistentLogicalTimestampEnabled() {
+ return getBooleanOrDefault(KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED);
+ }
+
public Boolean shouldAutoCommit() {
return getBoolean(AUTO_COMMIT_ENABLE);
}
@@ -1035,6 +1046,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;
}
@@ -1067,6 +1082,10 @@ public int getMinCommitsToKeep() {
return getInt(HoodieCompactionConfig.MIN_COMMITS_TO_KEEP);
}
+ public int getArchiveMergeFilesBatchSize() {
+ return getInt(HoodieCompactionConfig.ARCHIVE_MERGE_FILES_BATCH_SIZE);
+ }
+
public int getParquetSmallFileLimit() {
return getInt(HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT);
}
@@ -1095,6 +1114,18 @@ public boolean isAutoClean() {
return getBoolean(HoodieCompactionConfig.AUTO_CLEAN);
}
+ public boolean getArchiveMergeEnable() {
+ return getBoolean(HoodieCompactionConfig.ARCHIVE_MERGE_ENABLE);
+ }
+
+ public long getArchiveMergeSmallFileLimitBytes() {
+ return getLong(HoodieCompactionConfig.ARCHIVE_MERGE_SMALL_FILE_LIMIT_BYTES);
+ }
+
+ public boolean isAutoArchive() {
+ return getBoolean(HoodieCompactionConfig.AUTO_ARCHIVE);
+ }
+
public boolean isAsyncClean() {
return getBoolean(HoodieCompactionConfig.ASYNC_CLEAN);
}
@@ -1127,6 +1158,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 +1170,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);
}
@@ -1143,15 +1182,23 @@ public boolean isAsyncClusteringEnabled() {
return getBoolean(HoodieClusteringConfig.ASYNC_CLUSTERING_ENABLE);
}
- public boolean isPreserveHoodieCommitMetadata() {
+ public boolean isPreserveHoodieCommitMetadataForClustering() {
return getBoolean(HoodieClusteringConfig.PRESERVE_COMMIT_METADATA);
}
+ public boolean isPreserveHoodieCommitMetadataForCompaction() {
+ return getBoolean(HoodieCompactionConfig.PRESERVE_COMMIT_METADATA);
+ }
+
public boolean isClusteringEnabled() {
// TODO: future support async clustering
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 +1239,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);
}
@@ -1204,6 +1264,10 @@ public long getClusteringSmallFileLimit() {
return getLong(HoodieClusteringConfig.PLAN_STRATEGY_SMALL_FILE_LIMIT);
}
+ public String getClusteringPartitionFilterRegexPattern() {
+ return getString(HoodieClusteringConfig.PARTITION_REGEX_PATTERN);
+ }
+
public int getClusteringMaxNumGroups() {
return getInt(HoodieClusteringConfig.PLAN_STRATEGY_MAX_GROUPS);
}
@@ -1295,8 +1359,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 +1421,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 +1468,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 +1630,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 +1678,10 @@ public boolean getPushGatewayRandomJobNameSuffix() {
return getBoolean(HoodieMetricsPrometheusConfig.PUSHGATEWAY_RANDOM_JOBNAME_SUFFIX);
}
+ public String getMetricReporterMetricsNamePrefix() {
+ return getStringOrDefault(HoodieMetricsConfig.METRICS_REPORTER_PREFIX);
+ }
+
/**
* memory configs.
*/
@@ -1760,12 +1852,22 @@ public WriteConcurrencyMode getWriteConcurrencyMode() {
return WriteConcurrencyMode.fromValue(getString(WRITE_CONCURRENCY_MODE));
}
- public Boolean mergeDeltastreamerStateFromPreviousCommit() {
- return getBoolean(HoodieWriteConfig.WRITE_CONCURRENCY_MERGE_DELTASTREAMER_STATE);
+ /**
+ * Are any table services configured to run inline?
+ *
+ * @return True if any table services are configured to run inline, false otherwise.
+ */
+ public Boolean areAnyTableServicesInline() {
+ return inlineClusteringEnabled() || inlineCompactionEnabled() || isAutoClean();
}
- public Boolean inlineTableServices() {
- return inlineClusteringEnabled() || inlineCompactionEnabled() || isAutoClean();
+ /**
+ * Are any table services configured to run async?
+ *
+ * @return True if any table services are configured to run async, false otherwise.
+ */
+ public Boolean areAnyTableServicesAsync() {
+ return isAsyncClusteringEnabled() || !inlineCompactionEnabled() || isAsyncClean();
}
public String getPreCommitValidators() {
@@ -1796,6 +1898,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 +1926,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 +2157,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 +2228,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 +2269,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 +2281,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,12 +2298,26 @@ protected void setDefaults() {
writeConfig.setDefaultOnCondition(!isPayloadConfigSet,
HoodiePayloadConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
writeConfig.setDefaultOnCondition(!isMetadataConfigSet,
- HoodieMetadataConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
- writeConfig.setDefaultOnCondition(!isLockConfigSet,
- HoodieLockConfig.newBuilder().fromProperties(writeConfig.getProps()).build());
+ HoodieMetadataConfig.newBuilder().withEngineType(engineType).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));
+
+ // Async table services can update the metadata table and a lock provider is
+ // needed to guard against any concurrent table write operations. If user has
+ // not configured any lock provider, let's use the InProcess lock provider.
+ final TypedProperties writeConfigProperties = writeConfig.getProps();
+ final boolean isLockProviderPropertySet = writeConfigProperties.containsKey(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME)
+ || writeConfigProperties.containsKey(HoodieLockConfig.LOCK_PROVIDER_CLASS_PROP);
+ if (!isLockConfigSet) {
+ HoodieLockConfig.Builder lockConfigBuilder = HoodieLockConfig.newBuilder().fromProperties(writeConfig.getProps());
+ if (!isLockProviderPropertySet && writeConfig.areAnyTableServicesAsync()) {
+ lockConfigBuilder.withLockProvider(InProcessLockProvider.class);
+ }
+ writeConfig.setDefault(lockConfigBuilder.build());
+ }
}
private void validate() {
@@ -2201,5 +2338,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 60369d72fb4fc..8396540394b86 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 c04e8aa1e980e..e27ff1bcb0897 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 974f4d546b100..0428d12c40306 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/HoodieBaseBloomIndexCheckFunction.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
index 2ca563aa37ca3..441a212c59f40 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
@@ -49,8 +49,8 @@ public HoodieBaseBloomIndexCheckFunction(HoodieTable hoodieTable, HoodieWriteCon
}
@Override
- public Iterator> apply(Iterator> fileParitionRecordKeyTripletItr) {
- return new LazyKeyCheckIterator(fileParitionRecordKeyTripletItr);
+ public Iterator> apply(Iterator> filePartitionRecordKeyTripletItr) {
+ return new LazyKeyCheckIterator(filePartitionRecordKeyTripletItr);
}
class LazyKeyCheckIterator extends LazyIterableIterator, List> {
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 50d31f9cc50d2..18b094890081b 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 0000000000000..7dee9f3cdfa33
--- /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 0000000000000..acb06ea48bed1
--- /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 a33383a05c02d..6df05a7c6bd72 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;
@@ -162,7 +162,7 @@ private void init(HoodieRecord record) {
deltaWriteStat.setLogFiles(logFiles);
try {
- //save hoodie partition meta in the partition path
+ // Save hoodie partition meta in the partition path
HoodiePartitionMetadata partitionMetadata = new HoodiePartitionMetadata(fs, baseInstantTime,
new Path(config.getBasePath()), FSUtils.getPartitionPath(config.getBasePath(), partitionPath));
partitionMetadata.trySave(getPartitionId());
@@ -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 01ad45342c504..a9ff1f85478cb 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();
}
/**
@@ -139,6 +140,8 @@ public void write(HoodieRecord record, Option avroRecord) {
// Convert GenericRecord to GenericRecord with hoodie commit metadata in schema
IndexedRecord recordWithMetadataInSchema = rewriteRecord((GenericRecord) avroRecord.get());
if (preserveHoodieMetadata) {
+ // do not preserve FILENAME_METADATA_FIELD
+ recordWithMetadataInSchema.put(HoodieRecord.HOODIE_META_COLUMNS_NAME_TO_POS.get(HoodieRecord.FILENAME_METADATA_FIELD), path.getName());
fileWriter.writeAvro(record.getRecordKey(), recordWithMetadataInSchema);
} else {
fileWriter.writeAvroWithMetadata(recordWithMetadataInSchema, record);
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 b13b561f5dff1..87a8d133f0dd5 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
@@ -97,6 +97,7 @@ public class HoodieMergeHandle extends H
protected Map> keyToNewRecords;
protected Set writtenRecordKeys;
protected HoodieFileWriter fileWriter;
+ private boolean preserveMetadata = false;
protected Path newFilePath;
protected Path oldFilePath;
@@ -133,6 +134,7 @@ public HoodieMergeHandle(HoodieWriteConfig config, String instantTime, HoodieTab
super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier);
this.keyToNewRecords = keyToNewRecords;
this.useWriterSchema = true;
+ this.preserveMetadata = config.isPreserveHoodieCommitMetadataForCompaction();
init(fileId, this.partitionPath, dataFileToBeMerged);
validateAndSetAndKeyGenProps(keyGeneratorOpt, config.populateMetaFields());
}
@@ -250,11 +252,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 +272,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,14 +289,15 @@ 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 031f92cccdfaf..7e4c519a8fafc 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 6747c4a4a80e4..a719bcb8f334f 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 recordKeyFields) {
+ public static String getRecordKey(GenericRecord record, List recordKeyFields, boolean consistentLogicalTimestampEnabled) {
boolean keyIsNullEmpty = true;
StringBuilder recordKey = new StringBuilder();
for (String recordKeyField : recordKeyFields) {
- String recordKeyValue = HoodieAvroUtils.getNestedFieldValAsString(record, recordKeyField, true);
+ String recordKeyValue = HoodieAvroUtils.getNestedFieldValAsString(record, recordKeyField, true, consistentLogicalTimestampEnabled);
if (recordKeyValue == null) {
recordKey.append(recordKeyField + ":" + NULL_RECORDKEY_PLACEHOLDER + ",");
} else if (recordKeyValue.isEmpty()) {
@@ -112,14 +112,14 @@ public static String getRecordKey(GenericRecord record, List recordKeyFi
}
public static String getRecordPartitionPath(GenericRecord record, List partitionPathFields,
- boolean hiveStylePartitioning, boolean encodePartitionPath) {
+ boolean hiveStylePartitioning, boolean encodePartitionPath, boolean consistentLogicalTimestampEnabled) {
if (partitionPathFields.isEmpty()) {
return "";
}
StringBuilder partitionPath = new StringBuilder();
for (String partitionPathField : partitionPathFields) {
- String fieldVal = HoodieAvroUtils.getNestedFieldValAsString(record, partitionPathField, true);
+ String fieldVal = HoodieAvroUtils.getNestedFieldValAsString(record, partitionPathField, true, consistentLogicalTimestampEnabled);
if (fieldVal == null || fieldVal.isEmpty()) {
partitionPath.append(hiveStylePartitioning ? partitionPathField + "=" + HUDI_DEFAULT_PARTITION_PATH
: HUDI_DEFAULT_PARTITION_PATH);
@@ -135,8 +135,8 @@ public static String getRecordPartitionPath(GenericRecord record, List p
return partitionPath.toString();
}
- public static String getRecordKey(GenericRecord record, String recordKeyField) {
- String recordKey = HoodieAvroUtils.getNestedFieldValAsString(record, recordKeyField, true);
+ public static String getRecordKey(GenericRecord record, String recordKeyField, boolean consistentLogicalTimestampEnabled) {
+ String recordKey = HoodieAvroUtils.getNestedFieldValAsString(record, recordKeyField, true, consistentLogicalTimestampEnabled);
if (recordKey == null || recordKey.isEmpty()) {
throw new HoodieKeyException("recordKey value: \"" + recordKey + "\" for field: \"" + recordKeyField + "\" cannot be null or empty.");
}
@@ -144,8 +144,8 @@ public static String getRecordKey(GenericRecord record, String recordKeyField) {
}
public static String getPartitionPath(GenericRecord record, String partitionPathField,
- boolean hiveStylePartitioning, boolean encodePartitionPath) {
- String partitionPath = HoodieAvroUtils.getNestedFieldValAsString(record, partitionPathField, true);
+ boolean hiveStylePartitioning, boolean encodePartitionPath, boolean consistentLogicalTimestampEnabled) {
+ String partitionPath = HoodieAvroUtils.getNestedFieldValAsString(record, partitionPathField, true, consistentLogicalTimestampEnabled);
if (partitionPath == null || partitionPath.isEmpty()) {
partitionPath = HUDI_DEFAULT_PARTITION_PATH;
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/NonpartitionedAvroKeyGenerator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/NonpartitionedAvroKeyGenerator.java
index 4384f171dc96c..3f6aaadf6e803 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/NonpartitionedAvroKeyGenerator.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/NonpartitionedAvroKeyGenerator.java
@@ -57,9 +57,9 @@ public String getRecordKey(GenericRecord record) {
// 1. if there is only one record key field, the format of record key is just ""
// 2. if there are multiple record key fields, the format is ":,:,..."
if (getRecordKeyFieldNames().size() == 1) {
- return KeyGenUtils.getRecordKey(record, getRecordKeyFields().get(0));
+ return KeyGenUtils.getRecordKey(record, getRecordKeyFields().get(0), isConsistentLogicalTimestampEnabled());
}
- return KeyGenUtils.getRecordKey(record, getRecordKeyFields());
+ return KeyGenUtils.getRecordKey(record, getRecordKeyFields(), isConsistentLogicalTimestampEnabled());
}
public String getEmptyPartition() {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/SimpleAvroKeyGenerator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/SimpleAvroKeyGenerator.java
index dae55c5f8afce..943091225a3e1 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/SimpleAvroKeyGenerator.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/SimpleAvroKeyGenerator.java
@@ -47,11 +47,11 @@ public SimpleAvroKeyGenerator(TypedProperties props) {
@Override
public String getRecordKey(GenericRecord record) {
- return KeyGenUtils.getRecordKey(record, getRecordKeyFields().get(0));
+ return KeyGenUtils.getRecordKey(record, getRecordKeyFields().get(0), isConsistentLogicalTimestampEnabled());
}
@Override
public String getPartitionPath(GenericRecord record) {
- return KeyGenUtils.getPartitionPath(record, getPartitionPathFields().get(0), hiveStylePartitioning, encodePartitionPath);
+ return KeyGenUtils.getPartitionPath(record, getPartitionPathFields().get(0), hiveStylePartitioning, encodePartitionPath, isConsistentLogicalTimestampEnabled());
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/TimestampBasedAvroKeyGenerator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/TimestampBasedAvroKeyGenerator.java
index 430cde81d2e32..bc84ece503487 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/TimestampBasedAvroKeyGenerator.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/TimestampBasedAvroKeyGenerator.java
@@ -36,6 +36,7 @@
import java.io.IOException;
import java.io.Serializable;
import java.math.BigDecimal;
+import java.sql.Timestamp;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;
@@ -125,7 +126,7 @@ public TimestampBasedAvroKeyGenerator(TypedProperties config) throws IOException
@Override
public String getPartitionPath(GenericRecord record) {
- Object partitionVal = HoodieAvroUtils.getNestedFieldVal(record, getPartitionPathFields().get(0), true);
+ Object partitionVal = HoodieAvroUtils.getNestedFieldVal(record, getPartitionPathFields().get(0), true, isConsistentLogicalTimestampEnabled());
if (partitionVal == null) {
partitionVal = getDefaultPartitionVal();
}
@@ -191,6 +192,8 @@ public String getPartitionPath(Object partitionVal) {
timeMs = convertLongTimeToMillis(((Float) partitionVal).longValue());
} else if (partitionVal instanceof Long) {
timeMs = convertLongTimeToMillis((Long) partitionVal);
+ } else if (partitionVal instanceof Timestamp && isConsistentLogicalTimestampEnabled()) {
+ timeMs = ((Timestamp) partitionVal).getTime();
} else if (partitionVal instanceof Integer) {
timeMs = convertLongTimeToMillis(((Integer) partitionVal).longValue());
} else if (partitionVal instanceof BigDecimal) {
@@ -225,5 +228,4 @@ private long convertLongTimeToMillis(Long partitionVal) {
}
return MILLISECONDS.convert(partitionVal, timeUnit);
}
-
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
index 71d61938c7aa5..32f05cbad870c 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
@@ -41,10 +41,12 @@
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.marker.MarkerType;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
@@ -75,10 +77,12 @@
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import java.util.stream.Collectors;
import static org.apache.hudi.common.table.HoodieTableConfig.ARCHIVELOG_FOLDER;
import static org.apache.hudi.metadata.HoodieTableMetadata.METADATA_TABLE_NAME_SUFFIX;
+import static org.apache.hudi.metadata.HoodieTableMetadata.NON_PARTITIONED_NAME;
import static org.apache.hudi.metadata.HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP;
/**
@@ -90,6 +94,10 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
private static final Logger LOG = LogManager.getLogger(HoodieBackedTableMetadataWriter.class);
+ // Virtual keys support for metadata table. This Field is
+ // from the metadata payload schema.
+ private static final String RECORD_KEY_FIELD = HoodieMetadataPayload.SCHEMA_FIELD_ID_KEY;
+
protected HoodieWriteConfig metadataWriteConfig;
protected HoodieWriteConfig dataWriteConfig;
protected String tableName;
@@ -180,6 +188,8 @@ private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfi
.withAutoCommit(true)
.withAvroSchemaValidate(true)
.withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ .withRollbackUsingMarkers(false)
.withPath(HoodieTableMetadata.getMetadataTableBasePath(writeConfig.getBasePath()))
.withSchema(HoodieMetadataRecord.getClassSchema().toString())
.forTable(tableName)
@@ -194,12 +204,22 @@ private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfi
.archiveCommitsWith(minCommitsToKeep, maxCommitsToKeep)
// we will trigger compaction manually, to control the instant times
.withInlineCompaction(false)
- .withMaxNumDeltaCommitsBeforeCompaction(writeConfig.getMetadataCompactDeltaCommitMax()).build())
+ .withMaxNumDeltaCommitsBeforeCompaction(writeConfig.getMetadataCompactDeltaCommitMax())
+ // we will trigger archive manually, to ensure only regular writer invokes it
+ .withAutoArchive(false).build())
.withParallelism(parallelism, parallelism)
.withDeleteParallelism(parallelism)
.withRollbackParallelism(parallelism)
.withFinalizeWriteParallelism(parallelism)
- .withAllowMultiWriteOnSameInstant(true);
+ .withAllowMultiWriteOnSameInstant(true)
+ .withKeyGenerator(HoodieTableMetadataKeyGenerator.class.getCanonicalName())
+ .withPopulateMetaFields(dataWriteConfig.getMetadataConfig().populateMetaFields());
+
+ // RecordKey properties are needed for the metadata table records
+ final Properties properties = new Properties();
+ properties.put(HoodieTableConfig.RECORDKEY_FIELDS.key(), RECORD_KEY_FIELD);
+ properties.put("hoodie.datasource.write.recordkey.field", RECORD_KEY_FIELD);
+ builder.withProperties(properties);
if (writeConfig.isMetricsOn()) {
builder.withMetricsConfig(HoodieMetricsConfig.newBuilder()
@@ -224,6 +244,7 @@ private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfi
case PROMETHEUS_PUSHGATEWAY:
case CONSOLE:
case INMEMORY:
+ case CLOUDWATCH:
break;
default:
throw new HoodieMetadataException("Unsupported Metrics Reporter type " + writeConfig.getMetricsReporterType());
@@ -336,21 +357,11 @@ private 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