You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2021/12/04 09:40:24 UTC
[hudi] branch release-0.10.0 updated (92dbb41 -> e9e6939)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a change to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git.
from 92dbb41 Bumping release candidate number 2
new 69d1435 [MINOR] Fix typo,rename 'getUrlEncodePartitoning' to 'getUrlEncodePartitioning' (#4130)
new 63a7df7 [HUDI-2642] Add support ignoring case in update sql operation (#3882)
new c38ebbc [HUDI-2891] Fix write configs for Java engine in Kafka Connect Sink (#4161)
new df70e32 Revert "[HUDI-2855] Change the default value of 'PAYLOAD_CLASS_NAME' to 'DefaultHoodieRecordPayload' (#4115)" (#4169)
new 7df81a6 Revert "[HUDI-2856] Bit cask disk map delete modified (#4116)" (#4171)
new 9d7f681 [HUDI-2880] Fixing loading of props from default dir (#4167)
new 7d2c248 [HUDI-2881] Compact the file group with larger log files to reduce write amplification (#4152)
new 1ee35c4 Fixed partitions produced by layout optimization in case order-by key is composed of a single column (#4183)
new 3e2116e [MINOR] Fix the wrong usage of timestamp length variable bug (#4179)
new 318818b [HUDI-2904] Fix metadata table archival overstepping between regular writers and table services (#4186)
new 6c79a68 [HUDI-2914] Fix remote timeline server config for flink (#4191)
new fe2aae4 [minor] Refactor write profile to always generate fs view (#4198)
new 2a4e718 [HUDI-2924] Refresh the fs view on successful checkpoints for write profile (#4199)
new 1bf13d1 [MINOR] use catalog schema if can not find table schema (#4182)
new 7de5d17 [HUDI-2902] Fixing populate meta fields with Hfile writers and Disabling virtual keys by default for metadata table (#4194)
new 008df11 [HUDI-2911] Removing default value for `PARTITIONPATH_FIELD_NAME` resulting in incorrect `KeyGenerator` configuration (#4195)
new 342acbe Revert "[HUDI-2495] Resolve inconsistent key generation for timestamp types by GenericRecord and Row (#3944)" (#4201)
new a49e5ac [HUDI-2894][HUDI-2905] Metadata table - avoiding key lookup failures on base files over S3 (#4185)
new 474af05 Revert "[HUDI-2489]Tuning HoodieROTablePathFilter by caching hoodieTableFileSystemView, aiming to reduce unnecessary list/get requests"
new 31947a0 [MINOR] Mitigate CI jobs timeout issues (#4173)
new 7034d1a [HUDI-2933] DISABLE Metadata table by default (#4213)
new b88d4be [HUDI-2890] Kafka Connect: Fix failed writes and avoid table service concurrent operations (#4211)
new e131507 [HUDI-2923] Fixing metadata table reader when metadata compaction is inflight (#4206)
new 148e069 [HUDI-2934] Optimize RequestHandler code style
new 40879bc [HUDI-2935] Remove special casing of clustering in deltastreamer checkpoint retrival (#4216)
new e9e6939 Bumping release candidate number 3
The 26 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
azure-pipelines.yml | 14 +-
docker/hoodie/hadoop/base/pom.xml | 2 +-
docker/hoodie/hadoop/datanode/pom.xml | 2 +-
docker/hoodie/hadoop/historyserver/pom.xml | 2 +-
docker/hoodie/hadoop/hive_base/pom.xml | 2 +-
docker/hoodie/hadoop/namenode/pom.xml | 2 +-
docker/hoodie/hadoop/pom.xml | 2 +-
docker/hoodie/hadoop/prestobase/pom.xml | 2 +-
docker/hoodie/hadoop/spark_base/pom.xml | 2 +-
docker/hoodie/hadoop/sparkadhoc/pom.xml | 2 +-
docker/hoodie/hadoop/sparkmaster/pom.xml | 2 +-
docker/hoodie/hadoop/sparkworker/pom.xml | 2 +-
hudi-aws/pom.xml | 4 +-
hudi-cli/pom.xml | 2 +-
hudi-client/hudi-client-common/pom.xml | 4 +-
.../hudi/client/AbstractHoodieWriteClient.java | 40 ++-
.../apache/hudi/config/HoodieCompactionConfig.java | 23 ++
.../org/apache/hudi/config/HoodieWriteConfig.java | 12 +-
.../apache/hudi/io/HoodieSortedMergeHandle.java | 3 +-
.../hudi/io/storage/HoodieFileWriterFactory.java | 2 +-
.../apache/hudi/io/storage/HoodieHFileWriter.java | 14 +-
.../metadata/HoodieBackedTableMetadataWriter.java | 10 +-
.../LogFileSizeBasedCompactionStrategy.java | 9 +-
.../io/storage/TestHoodieHFileReaderWriter.java | 55 +++-
...chema.avsc => exampleSchemaWithMetaFields.avsc} | 36 ++-
hudi-client/hudi-flink-client/pom.xml | 4 +-
.../apache/hudi/client/HoodieFlinkWriteClient.java | 11 +-
.../FlinkHoodieBackedTableMetadataWriter.java | 3 +-
hudi-client/hudi-java-client/pom.xml | 4 +-
hudi-client/hudi-spark-client/pom.xml | 4 +-
.../SparkHoodieBackedTableMetadataWriter.java | 3 +-
.../java/org/apache/spark/OrderingIndexHelper.java | 2 +-
.../hudi/client/functional/TestHBaseIndex.java | 6 +-
.../functional/TestHoodieBackedMetadata.java | 281 +++++++++++++++++++++
.../client/functional/TestHoodieMetadataBase.java | 2 +-
.../strategy/TestHoodieCompactionStrategy.java | 9 +-
.../TestHoodieSparkMergeOnReadTableRollback.java | 2 +-
hudi-client/pom.xml | 2 +-
hudi-common/pom.xml | 2 +-
.../java/org/apache/hudi/avro/HoodieAvroUtils.java | 3 -
.../bootstrap/index/HFileBootstrapIndex.java | 8 +-
.../common/config/DFSPropertiesConfiguration.java | 31 ++-
.../hudi/common/config/HoodieMetadataConfig.java | 14 +-
.../hudi/common/table/HoodieTableConfig.java | 2 +-
.../common/util/collection/BitCaskDiskMap.java | 6 +-
...ruptedDataException.java => ExceptionUtil.java} | 22 +-
.../apache/hudi/io/storage/HoodieHFileReader.java | 2 +-
.../hudi/keygen/constant/KeyGeneratorOptions.java | 2 +-
.../hudi/metadata/HoodieBackedTableMetadata.java | 2 +-
.../HoodieMetadataMergedLogRecordReader.java | 4 +-
.../hudi/metadata/HoodieTableMetadataUtil.java | 8 +-
.../hudi/common/testutils/FileCreateUtils.java | 10 +
.../util/TestDFSPropertiesConfiguration.java | 5 +-
hudi-examples/pom.xml | 2 +-
hudi-flink/pom.xml | 2 +-
.../java/org/apache/hudi/sink/CleanFunction.java | 4 +-
.../sink/partitioner/BucketAssignFunction.java | 8 +-
.../hudi/sink/partitioner/BucketAssigner.java | 6 +-
.../partitioner/profile/DeltaWriteProfile.java | 12 +-
.../sink/partitioner/profile/WriteProfile.java | 51 ++--
.../java/org/apache/hudi/util/StreamerUtil.java | 21 +-
.../hudi/sink/partitioner/TestBucketAssigner.java | 2 +-
.../org/apache/hudi/utils/TestStreamerUtil.java | 11 +
hudi-hadoop-mr/pom.xml | 2 +-
.../hudi/hadoop/HoodieROTablePathFilter.java | 21 +-
hudi-integ-test/pom.xml | 2 +-
hudi-kafka-connect/README.md | 71 +++---
hudi-kafka-connect/demo/config-sink-hive.json | 1 -
hudi-kafka-connect/demo/config-sink.json | 1 -
hudi-kafka-connect/demo/setupKafka.sh | 2 +-
hudi-kafka-connect/pom.xml | 4 +-
.../transaction/ConnectTransactionCoordinator.java | 55 +++-
.../writers/ConnectTransactionServices.java | 2 +-
.../hudi/connect/writers/KafkaConnectConfigs.java | 14 +
.../writers/KafkaConnectTransactionServices.java | 30 ++-
.../writers/KafkaConnectWriterProvider.java | 14 +-
.../connect/TestConnectTransactionCoordinator.java | 117 ++++++---
.../helper/MockConnectTransactionServices.java | 3 +-
.../hudi/writers/TestBufferedConnectWriter.java | 2 +
.../test/resources/log4j-surefire-quiet.properties | 0
.../src/test/resources/log4j-surefire.properties | 0
hudi-spark-datasource/hudi-spark-common/pom.xml | 4 +-
.../scala/org/apache/hudi/DataSourceOptions.scala | 2 +-
hudi-spark-datasource/hudi-spark/pom.xml | 4 +-
.../scala/org/apache/hudi/HoodieWriterUtils.scala | 1 -
.../sql/catalyst/catalog/HoodieCatalogTable.scala | 2 +-
.../org/apache/spark/sql/hudi/HoodieSqlUtils.scala | 2 +-
.../hudi/command/DeleteHoodieTableCommand.scala | 2 +-
.../command/InsertIntoHoodieTableCommand.scala | 2 +-
.../hudi/command/MergeIntoHoodieTableCommand.scala | 2 +-
.../spark/sql/hudi/command/SqlKeyGenerator.scala | 3 +-
.../hudi/command/UpdateHoodieTableCommand.scala | 7 +-
.../hudi/TestGenericRecordAndRowConsistency.scala | 104 --------
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 33 ++-
.../apache/spark/sql/hudi/TestUpdateTable.scala | 45 +++-
hudi-spark-datasource/hudi-spark2/pom.xml | 4 +-
hudi-spark-datasource/hudi-spark3/pom.xml | 4 +-
hudi-spark-datasource/pom.xml | 2 +-
hudi-sync/hudi-dla-sync/pom.xml | 2 +-
hudi-sync/hudi-hive-sync/pom.xml | 2 +-
.../org/apache/hudi/hive/SchemaDifference.java | 1 -
.../hudi/hive/ddl/QueryBasedDDLExecutor.java | 5 +-
hudi-sync/hudi-sync-common/pom.xml | 2 +-
hudi-sync/pom.xml | 2 +-
hudi-timeline-service/pom.xml | 2 +-
.../hudi/timeline/service/RequestHandler.java | 14 +-
hudi-utilities/pom.xml | 2 +-
.../hudi/utilities/deltastreamer/DeltaSync.java | 3 -
...inuousModeWithMultipleWriters.COPY_ON_WRITE.zip | Bin 2592485 -> 2592484 bytes
...inuousModeWithMultipleWriters.MERGE_ON_READ.zip | Bin 3015940 -> 3015939 bytes
packaging/hudi-flink-bundle/pom.xml | 2 +-
packaging/hudi-hadoop-mr-bundle/pom.xml | 2 +-
packaging/hudi-hive-sync-bundle/pom.xml | 2 +-
packaging/hudi-integ-test-bundle/pom.xml | 2 +-
packaging/hudi-kafka-connect-bundle/pom.xml | 2 +-
packaging/hudi-presto-bundle/pom.xml | 2 +-
packaging/hudi-spark-bundle/pom.xml | 2 +-
packaging/hudi-timeline-server-bundle/pom.xml | 2 +-
packaging/hudi-utilities-bundle/pom.xml | 2 +-
pom.xml | 2 +-
120 files changed, 987 insertions(+), 455 deletions(-)
copy hudi-client/hudi-client-common/src/test/resources/{exampleEvolvedSchema.avsc => exampleSchemaWithMetaFields.avsc} (58%)
copy hudi-common/src/main/java/org/apache/hudi/exception/{HoodieCorruptedDataException.java => ExceptionUtil.java} (66%)
copy {hudi-utilities => hudi-kafka-connect}/src/test/resources/log4j-surefire-quiet.properties (100%)
copy {hudi-spark-datasource/hudi-spark3 => hudi-kafka-connect}/src/test/resources/log4j-surefire.properties (100%)
delete mode 100644 hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestGenericRecordAndRowConsistency.scala
[hudi] 04/26: Revert "[HUDI-2855] Change the default value of 'PAYLOAD_CLASS_NAME' to 'DefaultHoodieRecordPayload' (#4115)" (#4169)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit df70e32ae00f124d6de325ebf1fd3ccf434ed00a
Author: Alexey Kudinkin <al...@infinilake.com>
AuthorDate: Tue Nov 30 17:47:16 2021 -0800
Revert "[HUDI-2855] Change the default value of 'PAYLOAD_CLASS_NAME' to 'DefaultHoodieRecordPayload' (#4115)" (#4169)
This reverts commit 88067f57a23575aae3c371a7c7871e4207ca3bea.
(cherry picked from commit 24380c20606d63c7c129cb45a73f786f223e7d39)
---
.../src/main/java/org/apache/hudi/config/HoodieWriteConfig.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 f18339f..9ae22e8 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
@@ -32,7 +32,7 @@ import org.apache.hudi.common.fs.ConsistencyGuardConfig;
import org.apache.hudi.common.model.HoodieCleaningPolicy;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieFileFormat;
-import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
+import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.marker.MarkerType;
@@ -103,7 +103,7 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty<String> WRITE_PAYLOAD_CLASS_NAME = ConfigProperty
.key("hoodie.datasource.write.payload.class")
- .defaultValue(DefaultHoodieRecordPayload.class.getName())
+ .defaultValue(OverwriteWithLatestAvroPayload.class.getName())
.withDocumentation("Payload class used. Override this, if you like to roll your own merge logic, when upserting/inserting. "
+ "This will render any value set for PRECOMBINE_FIELD_OPT_VAL in-effective");
[hudi] 02/26: [HUDI-2642] Add support ignoring case in update sql operation (#3882)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 63a7df79e5e97159b39ae4354462b52d18899f22
Author: 董可伦 <do...@inspur.com>
AuthorDate: Tue Nov 30 14:36:36 2021 +0800
[HUDI-2642] Add support ignoring case in update sql operation (#3882)
(cherry picked from commit a398aad1fcbd5ff36eb7bc750fcc418d78b3a97e)
---
.../hudi/command/UpdateHoodieTableCommand.scala | 5 ++-
.../apache/spark/sql/hudi/TestUpdateTable.scala | 45 +++++++++++++++++++++-
2 files changed, 47 insertions(+), 3 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala
index 60122ce..7397b0d 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala
@@ -51,7 +51,10 @@ case class UpdateHoodieTableCommand(updateTable: UpdateTable) extends RunnableCo
}.toMap
val updateExpressions = table.output
- .map(attr => name2UpdateValue.getOrElse(attr.name, attr))
+ .map(attr => {
+ val UpdateValueOption = name2UpdateValue.find(f => sparkSession.sessionState.conf.resolver(f._1, attr.name))
+ if(UpdateValueOption.isEmpty) attr else UpdateValueOption.get._2
+ })
.filter { // filter the meta columns
case attr: AttributeReference =>
!HoodieRecord.HOODIE_META_COLUMNS.asScala.toSet.contains(attr.name)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestUpdateTable.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestUpdateTable.scala
index 2537c9c..82d067c 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestUpdateTable.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestUpdateTable.scala
@@ -27,6 +27,47 @@ class TestUpdateTable extends TestHoodieSqlBase {
spark.sql(
s"""
|create table $tableName (
+ | ID int,
+ | NAME string,
+ | PRICE double,
+ | TS long
+ |) using hudi
+ | location '${tmp.getCanonicalPath}/$tableName'
+ | options (
+ | type = '$tableType',
+ | primaryKey = 'ID',
+ | preCombineField = 'TS'
+ | )
+ """.stripMargin)
+ // insert data to table
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+ checkAnswer(s"select id, name, price, ts from $tableName")(
+ Seq(1, "a1", 10.0, 1000)
+ )
+
+ // update data
+ spark.sql(s"update $tableName set price = 20 where id = 1")
+ checkAnswer(s"select id, name, price, ts from $tableName")(
+ Seq(1, "a1", 20.0, 1000)
+ )
+
+ // update data
+ spark.sql(s"update $tableName set price = price * 2 where id = 1")
+ checkAnswer(s"select id, name, price, ts from $tableName")(
+ Seq(1, "a1", 40.0, 1000)
+ )
+ }
+ }
+ }
+
+ test("Test ignoring case for Update Table") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach {tableType =>
+ val tableName = generateTableName
+ // create table
+ spark.sql(
+ s"""
+ |create table $tableName (
| id int,
| name string,
| price double,
@@ -46,13 +87,13 @@ class TestUpdateTable extends TestHoodieSqlBase {
)
// update data
- spark.sql(s"update $tableName set price = 20 where id = 1")
+ spark.sql(s"update $tableName set PRICE = 20 where ID = 1")
checkAnswer(s"select id, name, price, ts from $tableName")(
Seq(1, "a1", 20.0, 1000)
)
// update data
- spark.sql(s"update $tableName set price = price * 2 where id = 1")
+ spark.sql(s"update $tableName set PRICE = PRICE * 2 where ID = 1")
checkAnswer(s"select id, name, price, ts from $tableName")(
Seq(1, "a1", 40.0, 1000)
)
[hudi] 19/26: Revert "[HUDI-2489]Tuning HoodieROTablePathFilter by caching hoodieTableFileSystemView, aiming to reduce unnecessary list/get requests"
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 474af05eb50af3943d4eedad9bafdf188d67f273
Author: zhangyue19921010 <69...@users.noreply.github.com>
AuthorDate: Sat Dec 4 10:56:53 2021 +0800
Revert "[HUDI-2489]Tuning HoodieROTablePathFilter by caching hoodieTableFileSystemView, aiming to reduce unnecessary list/get requests"
Co-authored-by: yuezhang <yu...@freewheel.tv>
(cherry picked from commit 5616830ae18ab3cbe1592d4e463a940342d25edc)
---
.../apache/hudi/hadoop/HoodieROTablePathFilter.java | 21 ++++++---------------
1 file changed, 6 insertions(+), 15 deletions(-)
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieROTablePathFilter.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieROTablePathFilter.java
index c797f59..d94018b 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieROTablePathFilter.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieROTablePathFilter.java
@@ -79,11 +79,6 @@ public class HoodieROTablePathFilter implements Configurable, PathFilter, Serial
Map<String, HoodieTableMetaClient> metaClientCache;
/**
- * HoodieTableFileSystemView Cache.
- */
- private Map<String, HoodieTableFileSystemView> hoodieTableFileSystemViewCache;
-
- /**
* Hadoop configurations for the FileSystem.
*/
private SerializableConfiguration conf;
@@ -102,7 +97,6 @@ public class HoodieROTablePathFilter implements Configurable, PathFilter, Serial
this.nonHoodiePathCache = new HashSet<>();
this.conf = new SerializableConfiguration(conf);
this.metaClientCache = new HashMap<>();
- this.hoodieTableFileSystemViewCache = new HashMap<>();
}
/**
@@ -181,15 +175,8 @@ public class HoodieROTablePathFilter implements Configurable, PathFilter, Serial
metaClientCache.put(baseDir.toString(), metaClient);
}
- HoodieTableMetaClient finalMetaClient = metaClient;
- fsView = hoodieTableFileSystemViewCache.computeIfAbsent(baseDir.toString(), key ->
- FileSystemViewManager.createInMemoryFileSystemView(
- engineContext,
- finalMetaClient,
- HoodieInputFormatUtils.buildMetadataConfig(getConf())
- )
- );
-
+ fsView = FileSystemViewManager.createInMemoryFileSystemView(engineContext,
+ metaClient, HoodieInputFormatUtils.buildMetadataConfig(getConf()));
String partition = FSUtils.getRelativePartitionPath(new Path(metaClient.getBasePath()), folder);
List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles(partition).collect(Collectors.toList());
// populate the cache
@@ -215,6 +202,10 @@ public class HoodieROTablePathFilter implements Configurable, PathFilter, Serial
}
nonHoodiePathCache.add(folder.toString());
return true;
+ } finally {
+ if (fsView != null) {
+ fsView.close();
+ }
}
} else {
// files is at < 3 level depth in FS tree, can't be hoodie dataset
[hudi] 16/26: [HUDI-2911] Removing default value for `PARTITIONPATH_FIELD_NAME` resulting in incorrect `KeyGenerator` configuration (#4195)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 008df11e196d0ee81aa22cdd64bd27ca11d121dd
Author: Alexey Kudinkin <al...@infinilake.com>
AuthorDate: Fri Dec 3 04:33:38 2021 -0800
[HUDI-2911] Removing default value for `PARTITIONPATH_FIELD_NAME` resulting in incorrect `KeyGenerator` configuration (#4195)
(cherry picked from commit bed7f9897a9127130c3d241df7634d44aa12167b)
---
.../org/apache/hudi/exception/ExceptionUtil.java | 42 ++++++++++++++++++++++
.../hudi/keygen/constant/KeyGeneratorOptions.java | 2 +-
.../scala/org/apache/hudi/DataSourceOptions.scala | 2 +-
.../scala/org/apache/hudi/HoodieWriterUtils.scala | 1 -
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 29 +++++++++++++--
5 files changed, 71 insertions(+), 5 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/exception/ExceptionUtil.java b/hudi-common/src/main/java/org/apache/hudi/exception/ExceptionUtil.java
new file mode 100644
index 0000000..a0550ba
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/exception/ExceptionUtil.java
@@ -0,0 +1,42 @@
+/*
+ * 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.exception;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Class collecting common utilities helping in handling {@link Exception}s
+ */
+public final class ExceptionUtil {
+ private ExceptionUtil() {}
+
+ /**
+ * Fetches inner-most cause of the provided {@link Throwable}
+ */
+ @Nonnull
+ public static Throwable getRootCause(@Nonnull Throwable t) {
+ Throwable cause = t;
+ while (cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+
+ return cause;
+ }
+
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/keygen/constant/KeyGeneratorOptions.java b/hudi-common/src/main/java/org/apache/hudi/keygen/constant/KeyGeneratorOptions.java
index 3e369a5..d8535a2 100644
--- a/hudi-common/src/main/java/org/apache/hudi/keygen/constant/KeyGeneratorOptions.java
+++ b/hudi-common/src/main/java/org/apache/hudi/keygen/constant/KeyGeneratorOptions.java
@@ -52,7 +52,7 @@ public class KeyGeneratorOptions extends HoodieConfig {
public static final ConfigProperty<String> PARTITIONPATH_FIELD_NAME = ConfigProperty
.key("hoodie.datasource.write.partitionpath.field")
- .defaultValue("partitionpath")
+ .noDefaultValue()
.withDocumentation("Partition path field. Value to be used at the partitionPath component of HoodieKey. "
+ "Actual value ontained by invoking .toString()");
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
index c1101f3..51bcd88 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
@@ -676,7 +676,7 @@ object DataSourceWriteOptions {
val PARTITIONPATH_FIELD_OPT_KEY = KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key()
/** @deprecated Use {@link PARTITIONPATH_FIELD} and its methods instead */
@Deprecated
- val DEFAULT_PARTITIONPATH_FIELD_OPT_VAL = PARTITIONPATH_FIELD.defaultValue()
+ val DEFAULT_PARTITIONPATH_FIELD_OPT_VAL = null
/** @deprecated Use {@link TABLE_NAME} and its methods instead */
@Deprecated
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala
index 52d4a52..282de54 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala
@@ -56,7 +56,6 @@ object HoodieWriterUtils {
hoodieConfig.setDefaultValue(PRECOMBINE_FIELD)
hoodieConfig.setDefaultValue(PAYLOAD_CLASS_NAME)
hoodieConfig.setDefaultValue(RECORDKEY_FIELD)
- hoodieConfig.setDefaultValue(PARTITIONPATH_FIELD)
hoodieConfig.setDefaultValue(KEYGENERATOR_CLASS_NAME)
hoodieConfig.setDefaultValue(ENABLE)
hoodieConfig.setDefaultValue(COMMIT_METADATA_KEYPREFIX)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
index 48f8408..fa248e4 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
@@ -25,8 +25,9 @@ import org.apache.hudi.common.config.HoodieConfig
import org.apache.hudi.common.model._
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient, TableSchemaResolver}
import org.apache.hudi.common.testutils.HoodieTestDataGenerator
+import org.apache.hudi.common.util.PartitionPathEncodeUtils
import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieWriteConfig}
-import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.exception.{ExceptionUtil, HoodieException}
import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode
import org.apache.hudi.functional.TestBootstrap
import org.apache.hudi.hive.HiveSyncConfig
@@ -40,13 +41,16 @@ import org.apache.spark.sql.hudi.command.SqlKeyGenerator
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SQLContext, SaveMode, SparkSession}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue, fail}
+import org.junit.jupiter.api.function.Executable
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.{CsvSource, EnumSource, ValueSource}
import org.mockito.ArgumentMatchers.any
import org.mockito.Mockito.{spy, times, verify}
+import org.scalatest.Assertions.assertThrows
import org.scalatest.Matchers.{assertResult, be, convertToAnyShouldWrapper, intercept}
+import java.io.IOException
import java.time.Instant
import java.util.{Collections, Date, UUID}
import scala.collection.JavaConversions._
@@ -346,7 +350,6 @@ class TestHoodieSparkSqlWriter {
@Test
def testInsertDatasetWithoutPrecombineField(): Unit = {
- //create a new table
val fooTableModifier = commonTableModifier.updated(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
.updated(DataSourceWriteOptions.INSERT_DROP_DUPS.key, "false")
@@ -376,6 +379,28 @@ class TestHoodieSparkSqlWriter {
}
/**
+ * Test case for insert dataset without partitioning field
+ */
+ @Test
+ def testInsertDatasetWithoutPartitionField(): Unit = {
+ val tableOpts =
+ commonTableModifier
+ .updated(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+
+ // generate the inserts
+ val schema = DataSourceTestUtils.getStructTypeExampleSchema
+ val structType = AvroConversionUtils.convertAvroSchemaToStructType(schema)
+ val records = DataSourceTestUtils.generateRandomRows(1)
+ val recordsSeq = convertRowListToSeq(records)
+ val df = spark.createDataFrame(sc.parallelize(recordsSeq), structType)
+
+ // try write to Hudi
+ assertThrows[IOException] {
+ HoodieSparkSqlWriter.write(sqlContext, SaveMode.Append, tableOpts - DataSourceWriteOptions.PARTITIONPATH_FIELD.key, df)
+ }
+ }
+
+ /**
* Test case for bulk insert dataset with datasource impl multiple rounds.
*/
@Test
[hudi] 17/26: Revert "[HUDI-2495] Resolve inconsistent key generation for timestamp types by GenericRecord and Row (#3944)" (#4201)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 342acbee33b0f76f7b3a3b328e2e988cc5fdef8f
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Sat Dec 4 00:13:38 2021 +0800
Revert "[HUDI-2495] Resolve inconsistent key generation for timestamp types by GenericRecord and Row (#3944)" (#4201)
(cherry picked from commit 2f96f4300b37207703b477979f2461bdd294ccf9)
---
.../java/org/apache/hudi/avro/HoodieAvroUtils.java | 3 -
.../spark/sql/hudi/command/SqlKeyGenerator.scala | 3 +-
.../hudi/TestGenericRecordAndRowConsistency.scala | 104 ---------------------
3 files changed, 1 insertion(+), 109 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
index ff8aefe..fdaa466 100644
--- a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
@@ -59,7 +59,6 @@ import java.io.OutputStream;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
-import java.sql.Timestamp;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Arrays;
@@ -542,8 +541,6 @@ public class HoodieAvroUtils {
private static Object convertValueForAvroLogicalTypes(Schema fieldSchema, Object fieldValue) {
if (fieldSchema.getLogicalType() == LogicalTypes.date()) {
return LocalDate.ofEpochDay(Long.parseLong(fieldValue.toString()));
- } else if (fieldSchema.getLogicalType() == LogicalTypes.timestampMicros()) {
- return new Timestamp(Long.parseLong(fieldValue.toString()) / 1000);
} else if (fieldSchema.getLogicalType() instanceof LogicalTypes.Decimal) {
Decimal dc = (Decimal) fieldSchema.getLogicalType();
DecimalConversion decimalConversion = new DecimalConversion();
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala
index 2cc3ece..e069df9 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.hudi.command
-import java.sql.Timestamp
import java.util.concurrent.TimeUnit.{MICROSECONDS, MILLISECONDS}
import org.apache.avro.generic.GenericRecord
@@ -97,7 +96,7 @@ class SqlKeyGenerator(props: TypedProperties) extends ComplexKeyGenerator(props)
val timeMs = if (rowType) { // In RowType, the partitionPathValue is the time format string, convert to millis
SqlKeyGenerator.sqlTimestampFormat.parseMillis(_partitionValue)
} else {
- Timestamp.valueOf(_partitionValue).getTime
+ MILLISECONDS.convert(_partitionValue.toLong, MICROSECONDS)
}
val timestampFormat = PartitionPathEncodeUtils.escapePathName(
SqlKeyGenerator.timestampTimeFormat.print(timeMs))
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestGenericRecordAndRowConsistency.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestGenericRecordAndRowConsistency.scala
deleted file mode 100644
index 985bf2e..0000000
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestGenericRecordAndRowConsistency.scala
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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
-
-import org.apache.hudi.config.HoodieWriteConfig
-import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
-import org.apache.spark.sql.DataFrame
-import org.junit.jupiter.api.Test
-
-import java.sql.{Date, Timestamp}
-
-class TestGenericRecordAndRowConsistency extends SparkClientFunctionalTestHarness {
-
- val commonOpts = Map(
- HoodieWriteConfig.TBL_NAME.key -> "hoodie_type_consistency_tbl",
- "hoodie.insert.shuffle.parallelism" -> "1",
- "hoodie.upsert.shuffle.parallelism" -> "1",
- DataSourceWriteOptions.TABLE_TYPE.key -> "COPY_ON_WRITE",
- DataSourceWriteOptions.RECORDKEY_FIELD.key -> "str,eventTime",
- DataSourceWriteOptions.PRECOMBINE_FIELD.key -> "typeId",
- DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "typeId",
- DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key -> "org.apache.hudi.keygen.ComplexKeyGenerator"
- )
-
- @Test
- def testTimestampTypeConsistency(): Unit = {
- val _spark = spark
- import _spark.implicits._
-
- val df = Seq(
- (1, Timestamp.valueOf("2014-01-01 23:00:01"), "abc"),
- (1, Timestamp.valueOf("2014-11-30 12:40:32"), "abc"),
- (2, Timestamp.valueOf("2016-12-29 09:54:00"), "def"),
- (2, Timestamp.valueOf("2016-05-09 10:12:43"), "def")
- ).toDF("typeId", "eventTime", "str")
-
- testConsistencyBetweenGenericRecordAndRow(df)
- }
-
- @Test
- def testDateTypeConsistency(): Unit = {
- val _spark = spark
- import _spark.implicits._
-
- val df = Seq(
- (1, Date.valueOf("2014-01-01"), "abc"),
- (1, Date.valueOf("2014-11-30"), "abc"),
- (2, Date.valueOf("2016-12-29"), "def"),
- (2, Date.valueOf("2016-05-09"), "def")
- ).toDF("typeId", "eventTime", "str")
-
- testConsistencyBetweenGenericRecordAndRow(df)
- }
-
- private def testConsistencyBetweenGenericRecordAndRow(df: DataFrame): Unit = {
- val _spark = spark
- import _spark.implicits._
-
- // upsert operation generate recordKey by GenericRecord
- val tempRecordPath = basePath + "/record_tbl/"
- df.write.format("hudi")
- .options(commonOpts)
- .option(DataSourceWriteOptions.OPERATION.key, "upsert")
- .mode(org.apache.spark.sql.SaveMode.Overwrite)
- .save(tempRecordPath)
-
- val data1 = spark.read.format("hudi")
- .load(tempRecordPath)
- .select("_hoodie_record_key")
- .map(_.toString()).collect().sorted
-
- // bulk_insert operation generate recordKey by Row
- val tempRowPath = basePath + "/row_tbl/"
- df.write.format("hudi")
- .options(commonOpts)
- .option(DataSourceWriteOptions.OPERATION.key, "bulk_insert")
- .mode(org.apache.spark.sql.SaveMode.Overwrite)
- .save(tempRowPath)
-
- val data2 = spark.read.format("hudi")
- .load(tempRowPath)
- .select("_hoodie_record_key")
- .map(_.toString()).collect().sorted
-
- assert(data1 sameElements data2)
- }
-
-}
[hudi] 13/26: [HUDI-2924] Refresh the fs view on successful checkpoints for write profile (#4199)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 2a4e7186883eb97eca7a5072844f47c77fbf1ef4
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Fri Dec 3 16:12:59 2021 +0800
[HUDI-2924] Refresh the fs view on successful checkpoints for write profile (#4199)
(cherry picked from commit 0699521f83eab843279ff7d78ac2b92fc80ad79b)
---
.../src/main/java/org/apache/hudi/sink/CleanFunction.java | 4 +++-
.../apache/hudi/sink/partitioner/BucketAssignFunction.java | 2 +-
.../hudi/sink/partitioner/profile/DeltaWriteProfile.java | 2 +-
.../apache/hudi/sink/partitioner/profile/WriteProfile.java | 9 ++++++++-
.../src/main/java/org/apache/hudi/util/StreamerUtil.java | 13 ++++++++++++-
5 files changed, 25 insertions(+), 5 deletions(-)
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
index 26ac9f3..13154b2 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
@@ -60,7 +60,9 @@ public class CleanFunction<T> extends AbstractRichFunction
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
- this.writeClient = StreamerUtil.createWriteClient(conf, getRuntimeContext());
+ // do not use the remote filesystem view because the async cleaning service
+ // local timeline is very probably to fall behind with the remote one.
+ this.writeClient = StreamerUtil.createWriteClient(conf, getRuntimeContext(), false);
this.executor = NonThrownExecutor.builder(LOG).build();
}
}
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
index 73fd668..14cad16 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
@@ -113,7 +113,7 @@ public class BucketAssignFunction<K, I, O extends HoodieRecord<?>>
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- HoodieWriteConfig writeConfig = StreamerUtil.getHoodieClientConfig(this.conf, true);
+ HoodieWriteConfig writeConfig = StreamerUtil.getHoodieClientConfig(this.conf, false);
HoodieFlinkEngineContext context = new HoodieFlinkEngineContext(
new SerializableConfiguration(StreamerUtil.getHadoopConf()),
new FlinkTaskContextSupplier(getRuntimeContext()));
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
index e1c890c..97b6b23 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
@@ -59,7 +59,7 @@ public class DeltaWriteProfile extends WriteProfile {
List<FileSlice> allSmallFileSlices = new ArrayList<>();
// If we can index log files, we can add more inserts to log files for fileIds including those under
// pending compaction.
- List<FileSlice> allFileSlices = getFileSystemView().getLatestFileSlicesBeforeOrOn(partitionPath, latestCommitTime.getTimestamp(), true)
+ List<FileSlice> allFileSlices = fsView.getLatestFileSlicesBeforeOrOn(partitionPath, latestCommitTime.getTimestamp(), true)
.collect(Collectors.toList());
for (FileSlice fileSlice : allFileSlices) {
if (isSmallFile(fileSlice)) {
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
index d13162f..f6840e5 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
@@ -94,6 +94,11 @@ public class WriteProfile {
private long reloadedCheckpointId;
/**
+ * The file system view cache for one checkpoint interval.
+ */
+ protected SyncableFileSystemView fsView;
+
+ /**
* Metadata cache to reduce IO of metadata files.
*/
private final Map<String, HoodieCommitMetadata> metadataCache;
@@ -111,6 +116,7 @@ public class WriteProfile {
this.recordsPerBucket = config.getCopyOnWriteInsertSplitSize();
this.metaClient = StreamerUtil.createMetaClient(config.getBasePath(), context.getHadoopConf().get());
this.metadataCache = new HashMap<>();
+ this.fsView = getFileSystemView();
// profile the record statistics on construction
recordProfile();
}
@@ -190,7 +196,7 @@ public class WriteProfile {
if (!commitTimeline.empty()) { // if we have some commits
HoodieInstant latestCommitTime = commitTimeline.lastInstant().get();
- List<HoodieBaseFile> allFiles = getFileSystemView()
+ List<HoodieBaseFile> allFiles = fsView
.getLatestBaseFilesBeforeOrOn(partitionPath, latestCommitTime.getTimestamp()).collect(Collectors.toList());
for (HoodieBaseFile file : allFiles) {
@@ -243,6 +249,7 @@ public class WriteProfile {
return;
}
this.metaClient.reloadActiveTimeline();
+ this.fsView = getFileSystemView();
recordProfile();
cleanMetadataCache(this.metaClient.getCommitsTimeline().filterCompletedInstants().getInstants());
this.smallFilesMap.clear();
diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
index 1e39c92..c417d7d 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
@@ -367,12 +367,23 @@ public class StreamerUtil {
*/
@SuppressWarnings("rawtypes")
public static HoodieFlinkWriteClient createWriteClient(Configuration conf, RuntimeContext runtimeContext) {
+ return createWriteClient(conf, runtimeContext, true);
+ }
+
+ /**
+ * Creates the Flink write client.
+ *
+ * <p>This expects to be used by client, set flag {@code loadFsViewStorageConfig} to use
+ * remote filesystem view storage config, or an in-memory filesystem view storage is used.
+ */
+ @SuppressWarnings("rawtypes")
+ public static HoodieFlinkWriteClient createWriteClient(Configuration conf, RuntimeContext runtimeContext, boolean loadFsViewStorageConfig) {
HoodieFlinkEngineContext context =
new HoodieFlinkEngineContext(
new SerializableConfiguration(getHadoopConf()),
new FlinkTaskContextSupplier(runtimeContext));
- HoodieWriteConfig writeConfig = getHoodieClientConfig(conf, true);
+ HoodieWriteConfig writeConfig = getHoodieClientConfig(conf, loadFsViewStorageConfig);
return new HoodieFlinkWriteClient<>(context, writeConfig);
}
[hudi] 09/26: [MINOR] Fix the wrong usage of timestamp length variable bug (#4179)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 3e2116e69fb33009ba3336720f2c7d8d00909232
Author: zzzhy <ca...@163.com>
AuthorDate: Thu Dec 2 22:47:31 2021 +0800
[MINOR] Fix the wrong usage of timestamp length variable bug (#4179)
Signed-off-by: zzzhy <ca...@163.com>
(cherry picked from commit 61a03bc072c8d446a7334fdf7a284fca0a91f8c0)
---
.../src/main/scala/org/apache/spark/sql/hudi/HoodieSqlUtils.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlUtils.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlUtils.scala
index 6bb9565..aa622e1 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlUtils.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlUtils.scala
@@ -299,7 +299,7 @@ object HoodieSqlUtils extends SparkAdapterSupport {
val instantLength = queryInstant.length
if (instantLength == 19 || instantLength == 23) { // for yyyy-MM-dd HH:mm:ss[.SSS]
HoodieInstantTimeGenerator.getInstantForDateString(queryInstant)
- } else if (instantLength == HoodieInstantTimeGenerator.SECS_INSTANT_TIMESTAMP_FORMAT
+ } else if (instantLength == HoodieInstantTimeGenerator.SECS_INSTANT_ID_LENGTH
|| instantLength == HoodieInstantTimeGenerator.MILLIS_INSTANT_ID_LENGTH) { // for yyyyMMddHHmmss[SSS]
HoodieActiveTimeline.parseDateFromInstantTime(queryInstant) // validate the format
queryInstant
[hudi] 11/26: [HUDI-2914] Fix remote timeline server config for flink (#4191)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 6c79a68ad91bac8f3bb0e4a97915e6ede90cfe6a
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Fri Dec 3 08:59:10 2021 +0800
[HUDI-2914] Fix remote timeline server config for flink (#4191)
(cherry picked from commit 934fe54cc57b508875383cd807735b1323fef754)
---
.../sink/partitioner/BucketAssignFunction.java | 6 ++---
.../hudi/sink/partitioner/BucketAssigner.java | 6 +----
.../partitioner/profile/DeltaWriteProfile.java | 12 ++++-----
.../sink/partitioner/profile/WriteProfile.java | 29 ++++++----------------
.../java/org/apache/hudi/util/StreamerUtil.java | 8 ++++--
.../hudi/sink/partitioner/TestBucketAssigner.java | 2 +-
.../org/apache/hudi/utils/TestStreamerUtil.java | 11 ++++++++
7 files changed, 34 insertions(+), 40 deletions(-)
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
index 33f1dd6..73fd668 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
@@ -121,16 +121,16 @@ public class BucketAssignFunction<K, I, O extends HoodieRecord<?>>
getRuntimeContext().getIndexOfThisSubtask(),
getRuntimeContext().getMaxNumberOfParallelSubtasks(),
getRuntimeContext().getNumberOfParallelSubtasks(),
- ignoreSmallFiles(writeConfig),
+ ignoreSmallFiles(),
HoodieTableType.valueOf(conf.getString(FlinkOptions.TABLE_TYPE)),
context,
writeConfig);
this.payloadCreation = PayloadCreation.instance(this.conf);
}
- private boolean ignoreSmallFiles(HoodieWriteConfig writeConfig) {
+ private boolean ignoreSmallFiles() {
WriteOperationType operationType = WriteOperationType.fromValue(conf.getString(FlinkOptions.OPERATION));
- return WriteOperationType.isOverwrite(operationType) || writeConfig.allowDuplicateInserts();
+ return WriteOperationType.isOverwrite(operationType);
}
@Override
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
index f9d5b1c..e73890f 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
@@ -22,7 +22,6 @@ import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.sink.partitioner.profile.WriteProfile;
import org.apache.hudi.sink.partitioner.profile.WriteProfiles;
-import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.commit.BucketInfo;
import org.apache.hudi.table.action.commit.BucketType;
import org.apache.hudi.table.action.commit.SmallFile;
@@ -188,6 +187,7 @@ public class BucketAssigner implements AutoCloseable {
smallFileAssignMap.put(partitionPath, assign);
return assign;
}
+ smallFileAssignMap.put(partitionPath, null);
return null;
}
@@ -211,10 +211,6 @@ public class BucketAssigner implements AutoCloseable {
this.writeProfile.reload(checkpointId);
}
- public HoodieTable<?, ?, ?, ?> getTable() {
- return this.writeProfile.getTable();
- }
-
private boolean fileIdOfThisTask(String fileId) {
// the file id can shuffle to this task
return KeyGroupRangeAssignment.assignKeyToParallelOperator(fileId, maxParallelism, numTasks) == taskID;
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
index 922c056..1f08c5a 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
@@ -25,7 +25,7 @@ import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
-import org.apache.hudi.common.table.view.AbstractTableFileSystemView;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.action.commit.SmallFile;
@@ -55,13 +55,11 @@ public class DeltaWriteProfile extends WriteProfile {
// Find out all eligible small file slices
if (!commitTimeline.empty()) {
HoodieInstant latestCommitTime = commitTimeline.lastInstant().get();
- // initialize the filesystem view based on the commit metadata
- initFileSystemView();
- // find smallest file in partition and append to it
+ // find the smallest file in partition and append to it
List<FileSlice> allSmallFileSlices = new ArrayList<>();
// If we can index log files, we can add more inserts to log files for fileIds including those under
// pending compaction.
- List<FileSlice> allFileSlices = fsView.getLatestFileSlicesBeforeOrOn(partitionPath, latestCommitTime.getTimestamp(), true)
+ List<FileSlice> allFileSlices = getFileSystemView().getLatestFileSlicesBeforeOrOn(partitionPath, latestCommitTime.getTimestamp(), true)
.collect(Collectors.toList());
for (FileSlice fileSlice : allFileSlices) {
if (isSmallFile(fileSlice)) {
@@ -91,8 +89,8 @@ public class DeltaWriteProfile extends WriteProfile {
return smallFileLocations;
}
- protected AbstractTableFileSystemView getFileSystemView() {
- return (AbstractTableFileSystemView) this.table.getSliceView();
+ protected SyncableFileSystemView getFileSystemView() {
+ return (SyncableFileSystemView) this.table.getSliceView();
}
private long getTotalFileSize(FileSlice fileSlice) {
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
index 1171a54..98eb29c 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
@@ -23,9 +23,10 @@ import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
-import org.apache.hudi.common.table.view.AbstractTableFileSystemView;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.sink.partitioner.BucketAssigner;
import org.apache.hudi.table.HoodieFlinkTable;
@@ -92,11 +93,6 @@ public class WriteProfile {
private long reloadedCheckpointId;
/**
- * The file system view cache for one checkpoint interval.
- */
- protected AbstractTableFileSystemView fsView;
-
- /**
* Metadata cache to reduce IO of metadata files.
*/
private final Map<String, HoodieCommitMetadata> metadataCache;
@@ -120,8 +116,8 @@ public class WriteProfile {
return recordsPerBucket;
}
- public HoodieTable<?, ?, ?, ?> getTable() {
- return table;
+ public HoodieTableMetaClient getMetaClient() {
+ return this.table.getMetaClient();
}
/**
@@ -183,9 +179,7 @@ public class WriteProfile {
if (!commitTimeline.empty()) { // if we have some commits
HoodieInstant latestCommitTime = commitTimeline.lastInstant().get();
- // initialize the filesystem view based on the commit metadata
- initFileSystemView();
- List<HoodieBaseFile> allFiles = fsView
+ List<HoodieBaseFile> allFiles = getFileSystemView()
.getLatestBaseFilesBeforeOrOn(partitionPath, latestCommitTime.getTimestamp()).collect(Collectors.toList());
for (HoodieBaseFile file : allFiles) {
@@ -203,15 +197,8 @@ public class WriteProfile {
return smallFileLocations;
}
- @VisibleForTesting
- public void initFileSystemView() {
- if (fsView == null) {
- fsView = getFileSystemView();
- }
- }
-
- protected AbstractTableFileSystemView getFileSystemView() {
- return (AbstractTableFileSystemView) this.table.getBaseFileOnlyView();
+ protected SyncableFileSystemView getFileSystemView() {
+ return (SyncableFileSystemView) HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) table.getContext()).getBaseFileOnlyView();
}
/**
@@ -245,9 +232,7 @@ public class WriteProfile {
return;
}
this.table.getMetaClient().reloadActiveTimeline();
- this.table.getHoodieView().sync();
recordProfile();
- this.fsView = null;
cleanMetadataCache(this.table.getMetaClient().getCommitsTimeline().filterCompletedInstants().getInstants());
this.smallFilesMap.clear();
this.reloadedCheckpointId = checkpointId;
diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
index 28a70b9..1e39c92 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
@@ -365,6 +365,7 @@ public class StreamerUtil {
*
* <p>This expects to be used by client, the driver should start an embedded timeline server.
*/
+ @SuppressWarnings("rawtypes")
public static HoodieFlinkWriteClient createWriteClient(Configuration conf, RuntimeContext runtimeContext) {
HoodieFlinkEngineContext context =
new HoodieFlinkEngineContext(
@@ -382,17 +383,20 @@ public class StreamerUtil {
*
* <p>The task context supplier is a constant: the write token is always '0-1-0'.
*/
+ @SuppressWarnings("rawtypes")
public static HoodieFlinkWriteClient createWriteClient(Configuration conf) throws IOException {
HoodieWriteConfig writeConfig = getHoodieClientConfig(conf, true, false);
+ // build the write client to start the embedded timeline server
+ final HoodieFlinkWriteClient writeClient = new HoodieFlinkWriteClient<>(HoodieFlinkEngineContext.DEFAULT, writeConfig);
// create the filesystem view storage properties for client
- FileSystemViewStorageConfig viewStorageConfig = writeConfig.getViewStorageConfig();
+ final FileSystemViewStorageConfig viewStorageConfig = writeConfig.getViewStorageConfig();
// rebuild the view storage config with simplified options.
FileSystemViewStorageConfig rebuilt = FileSystemViewStorageConfig.newBuilder()
.withStorageType(viewStorageConfig.getStorageType())
.withRemoteServerHost(viewStorageConfig.getRemoteViewServerHost())
.withRemoteServerPort(viewStorageConfig.getRemoteViewServerPort()).build();
ViewStorageProperties.createProperties(conf.getString(FlinkOptions.PATH), rebuilt);
- return new HoodieFlinkWriteClient<>(HoodieFlinkEngineContext.DEFAULT, writeConfig);
+ return writeClient;
}
/**
diff --git a/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestBucketAssigner.java b/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestBucketAssigner.java
index 053c2a3..4f4b549 100644
--- a/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestBucketAssigner.java
+++ b/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestBucketAssigner.java
@@ -401,7 +401,7 @@ public class TestBucketAssigner {
}
private static String getLastCompleteInstant(WriteProfile profile) {
- return StreamerUtil.getLastCompletedInstant(profile.getTable().getMetaClient());
+ return StreamerUtil.getLastCompletedInstant(profile.getMetaClient());
}
private void assertBucketEquals(
diff --git a/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java b/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
index e00fbfa..57297c5 100644
--- a/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
+++ b/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
@@ -19,9 +19,12 @@
package org.apache.hudi.utils;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
+import org.apache.hudi.common.table.view.FileSystemViewStorageType;
import org.apache.hudi.common.util.FileIOUtils;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.util.ViewStorageProperties;
import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.Test;
@@ -98,5 +101,13 @@ public class TestStreamerUtil {
long diff = StreamerUtil.instantTimeDiffSeconds(higher, lower);
assertThat(diff, is(75L));
}
+
+ @Test
+ void testDumpRemoteViewStorageConfig() throws IOException {
+ Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ StreamerUtil.createWriteClient(conf);
+ FileSystemViewStorageConfig storageConfig = ViewStorageProperties.loadFromProperties(conf.getString(FlinkOptions.PATH));
+ assertThat(storageConfig.getStorageType(), is(FileSystemViewStorageType.REMOTE_FIRST));
+ }
}
[hudi] 01/26: [MINOR] Fix typo, rename 'getUrlEncodePartitoning' to 'getUrlEncodePartitioning' (#4130)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 69d14353bcc20c07d53b7835940c4117cfbf5110
Author: 董可伦 <do...@inspur.com>
AuthorDate: Tue Nov 30 10:31:22 2021 +0800
[MINOR] Fix typo,rename 'getUrlEncodePartitoning' to 'getUrlEncodePartitioning' (#4130)
(cherry picked from commit 3433f00cb5366fa190a7b2d5254255041a705deb)
---
.../src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java | 2 +-
.../org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala | 2 +-
.../apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala | 2 +-
.../apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala | 2 +-
.../org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala | 2 +-
.../src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala | 4 ++--
6 files changed, 7 insertions(+), 7 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
index 1da4130..e4b60e2 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
@@ -466,7 +466,7 @@ public class HoodieTableConfig extends HoodieConfig {
return getString(HIVE_STYLE_PARTITIONING_ENABLE);
}
- public String getUrlEncodePartitoning() {
+ public String getUrlEncodePartitioning() {
return getString(URL_ENCODE_PARTITIONING);
}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala
index 1decb5d..8475b63 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala
@@ -72,7 +72,7 @@ case class DeleteHoodieTableCommand(deleteTable: DeleteFromTable) extends Runnab
"path" -> path,
TBL_NAME.key -> tableConfig.getTableName,
HIVE_STYLE_PARTITIONING.key -> tableConfig.getHiveStylePartitioningEnable,
- URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitoning,
+ URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitioning,
KEYGENERATOR_CLASS_NAME.key -> classOf[SqlKeyGenerator].getCanonicalName,
SqlKeyGenerator.ORIGIN_KEYGEN_CLASS_NAME -> tableConfig.getKeyGeneratorClassName,
OPERATION.key -> DataSourceWriteOptions.DELETE_OPERATION_OPT_VAL,
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
index 635aa64..4b6d8e0 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
@@ -212,7 +212,7 @@ object InsertIntoHoodieTableCommand extends Logging {
val partitionFields = hoodieCatalogTable.partitionFields.mkString(",")
val hiveStylePartitioningEnable = Option(tableConfig.getHiveStylePartitioningEnable).getOrElse("true")
- val urlEncodePartitioning = Option(tableConfig.getUrlEncodePartitoning).getOrElse("false")
+ val urlEncodePartitioning = Option(tableConfig.getUrlEncodePartitioning).getOrElse("false")
val keyGeneratorClassName = Option(tableConfig.getKeyGeneratorClassName)
.getOrElse(classOf[ComplexKeyGenerator].getCanonicalName)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala
index a746f83..76c8715 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala
@@ -449,7 +449,7 @@ case class MergeIntoHoodieTableCommand(mergeInto: MergeIntoTable) extends Runnab
PARTITIONPATH_FIELD.key -> tableConfig.getPartitionFieldProp,
PAYLOAD_CLASS_NAME.key -> classOf[ExpressionPayload].getCanonicalName,
HIVE_STYLE_PARTITIONING.key -> tableConfig.getHiveStylePartitioningEnable,
- URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitoning,
+ URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitioning,
KEYGENERATOR_CLASS_NAME.key -> classOf[SqlKeyGenerator].getCanonicalName,
SqlKeyGenerator.ORIGIN_KEYGEN_CLASS_NAME -> tableConfig.getKeyGeneratorClassName,
META_SYNC_ENABLED.key -> enableHive.toString,
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala
index a6a0cb1..60122ce 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/UpdateHoodieTableCommand.scala
@@ -98,7 +98,7 @@ case class UpdateHoodieTableCommand(updateTable: UpdateTable) extends RunnableCo
PRECOMBINE_FIELD.key -> preCombineColumn,
TBL_NAME.key -> hoodieCatalogTable.tableName,
HIVE_STYLE_PARTITIONING.key -> tableConfig.getHiveStylePartitioningEnable,
- URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitoning,
+ URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitioning,
KEYGENERATOR_CLASS_NAME.key -> classOf[SqlKeyGenerator].getCanonicalName,
SqlKeyGenerator.ORIGIN_KEYGEN_CLASS_NAME -> tableConfig.getKeyGeneratorClassName,
OPERATION.key -> UPSERT_OPERATION_OPT_VAL,
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
index fbdfb69..48f8408 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
@@ -853,7 +853,7 @@ class TestHoodieSparkSqlWriter {
.setConf(spark.sparkContext.hadoopConfiguration)
.setBasePath(tablePath1).build().getTableConfig
assert(tableConfig1.getHiveStylePartitioningEnable == "true")
- assert(tableConfig1.getUrlEncodePartitoning == "false")
+ assert(tableConfig1.getUrlEncodePartitioning == "false")
assert(tableConfig1.getKeyGeneratorClassName == classOf[ComplexKeyGenerator].getName)
df.write.format("hudi")
.options(options)
@@ -874,7 +874,7 @@ class TestHoodieSparkSqlWriter {
.setConf(spark.sparkContext.hadoopConfiguration)
.setBasePath(tablePath2).build().getTableConfig
assert(tableConfig2.getHiveStylePartitioningEnable == "false")
- assert(tableConfig2.getUrlEncodePartitoning == "true")
+ assert(tableConfig2.getUrlEncodePartitioning == "true")
assert(tableConfig2.getKeyGeneratorClassName == classOf[SimpleKeyGenerator].getName)
val df2 = Seq((2, "a2", 20, 1000, "2021-10-16")).toDF("id", "name", "value", "ts", "dt")
[hudi] 24/26: [HUDI-2934] Optimize RequestHandler code style
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 148e069f31f9b2166c3833a4103deeb1cc7e43b3
Author: fengli <ld...@163.com>
AuthorDate: Sat Dec 4 13:56:25 2021 +0800
[HUDI-2934] Optimize RequestHandler code style
close apache/hudi#4215
(cherry picked from commit 568181a3e7c1d43a4ee8573d3746600be4f94379)
---
.../main/java/org/apache/hudi/hive/SchemaDifference.java | 1 -
.../org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java | 5 +++--
.../org/apache/hudi/timeline/service/RequestHandler.java | 14 +++++++-------
3 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/SchemaDifference.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/SchemaDifference.java
index e207b85..f48208a 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/SchemaDifference.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/SchemaDifference.java
@@ -28,7 +28,6 @@ import java.util.List;
import java.util.Map;
import java.util.StringJoiner;
-
/**
* Represents the schema difference between the storage schema and hive table schema.
*/
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java
index 6fbcfa9..aed2bbe 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java
@@ -39,8 +39,9 @@ import java.util.List;
import java.util.Map;
import static org.apache.hudi.hive.util.HiveSchemaUtil.HIVE_ESCAPE_CHARACTER;
-/*
-This class adds functionality for all query based DDLExecutors. The classes extending it only have to provide runSQL(sql) functions.
+
+/**
+ * This class adds functionality for all query based DDLExecutors. The classes extending it only have to provide runSQL(sql) functions.
*/
public abstract class QueryBasedDDLExecutor implements DDLExecutor {
private static final Logger LOG = LogManager.getLogger(QueryBasedDDLExecutor.class);
diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
index c4edc18..4744fbb 100644
--- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
+++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
@@ -500,14 +500,14 @@ public class RequestHandler {
if (refreshCheck) {
long beginFinalCheck = System.currentTimeMillis();
- String errMsg =
- "Last known instant from client was "
- + context.queryParam(RemoteHoodieTableFileSystemView.LAST_INSTANT_TS,
- HoodieTimeline.INVALID_INSTANT_TS)
- + " but server has the following timeline "
- + viewManager.getFileSystemView(context.queryParam(RemoteHoodieTableFileSystemView.BASEPATH_PARAM))
- .getTimeline().getInstants().collect(Collectors.toList());
if (isLocalViewBehind(context)) {
+ String errMsg =
+ "Last known instant from client was "
+ + context.queryParam(RemoteHoodieTableFileSystemView.LAST_INSTANT_TS,
+ HoodieTimeline.INVALID_INSTANT_TS)
+ + " but server has the following timeline "
+ + viewManager.getFileSystemView(context.queryParam(RemoteHoodieTableFileSystemView.BASEPATH_PARAM))
+ .getTimeline().getInstants().collect(Collectors.toList());
throw new BadRequestResponse(errMsg);
}
long endFinalCheck = System.currentTimeMillis();
[hudi] 23/26: [HUDI-2923] Fixing metadata table reader when metadata compaction is inflight (#4206)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e131507d4672316b6a0f2b2f3f63d8ad34e49b85
Author: Sivabalan Narayanan <si...@uber.com>
AuthorDate: Sat Dec 4 00:44:50 2021 -0500
[HUDI-2923] Fixing metadata table reader when metadata compaction is inflight (#4206)
* [HUDI-2923] Fixing metadata table reader when metadata compaction is inflight
* Fixing retry of pending compaction in metadata table and enhancing tests
(cherry picked from commit 1d4fb827e73b2ae510f0e2c6510a448c0b5bd5b3)
---
.../hudi/client/AbstractHoodieWriteClient.java | 7 ++
.../metadata/HoodieBackedTableMetadataWriter.java | 6 +-
.../FlinkHoodieBackedTableMetadataWriter.java | 2 +-
.../SparkHoodieBackedTableMetadataWriter.java | 2 +-
.../functional/TestHoodieBackedMetadata.java | 86 ++++++++++++++++++++++
.../hudi/metadata/HoodieBackedTableMetadata.java | 2 +-
.../hudi/metadata/HoodieTableMetadataUtil.java | 8 +-
.../hudi/common/testutils/FileCreateUtils.java | 10 +++
8 files changed, 117 insertions(+), 6 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
index 59acbb2..5c2bee1 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
@@ -520,6 +520,13 @@ public abstract class AbstractHoodieWriteClient<T extends HoodieRecordPayload, I
}
/**
+ * Run any pending compactions.
+ */
+ public void runAnyPendingCompactions() {
+ runAnyPendingCompactions(createTable(config, hadoopConf, config.isMetadataTableEnabled()));
+ }
+
+ /**
* Create a savepoint based on the latest commit action on the timeline.
*
* @param user - User creating the savepoint
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 54284fc..eb4b24a 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
@@ -682,7 +682,10 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
* deltacommit.
*/
protected void compactIfNecessary(AbstractHoodieWriteClient writeClient, String instantTime) {
- String latestDeltacommitTime = metadataMetaClient.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().lastInstant()
+ // finish off any pending compactions if any from previous attempt.
+ writeClient.runAnyPendingCompactions();
+
+ String latestDeltacommitTime = metadataMetaClient.reloadActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().lastInstant()
.get().getTimestamp();
List<HoodieInstant> pendingInstants = dataMetaClient.reloadActiveTimeline().filterInflightsAndRequested()
.findInstantsBefore(latestDeltacommitTime).getInstants().collect(Collectors.toList());
@@ -693,6 +696,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
return;
}
+
// Trigger compaction with suffixes based on the same instant time. This ensures that any future
// delta commits synced over will not have an instant time lesser than the last completed instant on the
// metadata table.
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
index 0dcfcfc..d11f570 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
@@ -154,7 +154,7 @@ public class FlinkHoodieBackedTableMetadataWriter extends HoodieBackedTableMetad
* The record is tagged with respective file slice's location based on its record key.
*/
private List<HoodieRecord> prepRecords(List<HoodieRecord> records, String partitionName, int numFileGroups) {
- List<FileSlice> fileSlices = HoodieTableMetadataUtil.loadPartitionFileGroupsWithLatestFileSlices(metadataMetaClient, partitionName);
+ List<FileSlice> fileSlices = HoodieTableMetadataUtil.loadPartitionFileGroupsWithLatestFileSlices(metadataMetaClient, partitionName, false);
ValidationUtils.checkArgument(fileSlices.size() == numFileGroups, String.format("Invalid number of file groups: found=%d, required=%d", fileSlices.size(), numFileGroups));
return records.stream().map(r -> {
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
index 65ade82..b7b5961 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
@@ -169,7 +169,7 @@ public class SparkHoodieBackedTableMetadataWriter extends HoodieBackedTableMetad
* The record is tagged with respective file slice's location based on its record key.
*/
private JavaRDD<HoodieRecord> prepRecords(JavaRDD<HoodieRecord> recordsRDD, String partitionName, int numFileGroups) {
- List<FileSlice> fileSlices = HoodieTableMetadataUtil.loadPartitionFileGroupsWithLatestFileSlices(metadataMetaClient, partitionName);
+ List<FileSlice> fileSlices = HoodieTableMetadataUtil.loadPartitionFileGroupsWithLatestFileSlices(metadataMetaClient, partitionName, false);
ValidationUtils.checkArgument(fileSlices.size() == numFileGroups, String.format("Invalid number of file groups: found=%d, required=%d", fileSlices.size(), numFileGroups));
return recordsRDD.map(r -> {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
index ed245da..0339c47 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
@@ -54,6 +54,7 @@ import org.apache.hudi.common.table.view.FileSystemViewStorageType;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.table.view.TableFileSystemView;
import org.apache.hudi.common.testutils.FileCreateUtils;
+import org.apache.hudi.common.testutils.HoodieMetadataTestTable;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestTable;
import org.apache.hudi.common.util.HoodieTimer;
@@ -413,6 +414,91 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
});
}
+
+ /**
+ * Tests that virtual key configs are honored in base files after compaction in metadata table.
+ *
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testMetadataTableWithPendingCompaction(boolean simulateFailedCompaction) throws Exception {
+ HoodieTableType tableType = COPY_ON_WRITE;
+ init(tableType, false);
+ writeConfig = getWriteConfigBuilder(true, true, false)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .enableFullScan(true)
+ .enableMetrics(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(3)
+ .build()).build();
+ initWriteConfigAndMetatableWriter(writeConfig, true);
+
+ doWriteOperation(testTable, "0000001", INSERT);
+ // create an inflight compaction in metadata table.
+ // not easy to create an inflight in metadata table directly, hence letting compaction succeed and then deleting the completed instant.
+ // this new write is expected to trigger metadata table compaction
+ String commitInstant = "0000002";
+ doWriteOperation(testTable, commitInstant, INSERT);
+
+ HoodieTableMetadata tableMetadata = metadata(writeConfig, context);
+ String metadataCompactionInstant = commitInstant + "001";
+ assertTrue(tableMetadata.getLatestCompactionTime().isPresent());
+ assertEquals(tableMetadata.getLatestCompactionTime().get(), metadataCompactionInstant);
+
+ validateMetadata(testTable);
+ // Fetch compaction Commit file and rename to some other file. completed compaction meta file should have some serialized info that table interprets
+ // for future upserts. so, renaming the file here to some temp name and later renaming it back to same name.
+ java.nio.file.Path parentPath = Paths.get(metadataTableBasePath, HoodieTableMetaClient.METAFOLDER_NAME);
+ java.nio.file.Path metaFilePath = parentPath.resolve(metadataCompactionInstant + HoodieTimeline.COMMIT_EXTENSION);
+ java.nio.file.Path tempFilePath = FileCreateUtils.renameFileToTemp(metaFilePath, metadataCompactionInstant);
+ metaClient.reloadActiveTimeline();
+ testTable = HoodieMetadataTestTable.of(metaClient, metadataWriter);
+ // this validation will exercise the code path where a compaction is inflight in metadata table, but still metadata based file listing should match non
+ // metadata based file listing.
+ validateMetadata(testTable);
+
+ if (simulateFailedCompaction) {
+ // this should retry the compaction in metadata table.
+ doWriteOperation(testTable, "0000003", INSERT);
+ } else {
+ // let the compaction succeed in metadata and validation should succeed.
+ FileCreateUtils.renameTempToMetaFile(tempFilePath, metaFilePath);
+ }
+
+ validateMetadata(testTable);
+
+ // add few more write and validate
+ doWriteOperation(testTable, "0000004", INSERT);
+ doWriteOperation(testTable, "0000005", UPSERT);
+ validateMetadata(testTable);
+
+ if (simulateFailedCompaction) {
+ //trigger another compaction failure.
+ metadataCompactionInstant = "0000005001";
+ tableMetadata = metadata(writeConfig, context);
+ assertTrue(tableMetadata.getLatestCompactionTime().isPresent());
+ assertEquals(tableMetadata.getLatestCompactionTime().get(), metadataCompactionInstant);
+
+ // Fetch compaction Commit file and rename to some other file. completed compaction meta file should have some serialized info that table interprets
+ // for future upserts. so, renaming the file here to some temp name and later renaming it back to same name.
+ parentPath = Paths.get(metadataTableBasePath, HoodieTableMetaClient.METAFOLDER_NAME);
+ metaFilePath = parentPath.resolve(metadataCompactionInstant + HoodieTimeline.COMMIT_EXTENSION);
+ tempFilePath = FileCreateUtils.renameFileToTemp(metaFilePath, metadataCompactionInstant);
+
+ validateMetadata(testTable);
+
+ // this should retry the failed compaction in metadata table.
+ doWriteOperation(testTable, "0000006", INSERT);
+
+ validateMetadata(testTable);
+
+ // add few more write and validate
+ doWriteOperation(testTable, "0000007", INSERT);
+ doWriteOperation(testTable, "0000008", UPSERT);
+ validateMetadata(testTable);
+ }
+ }
+
/**
* Test rollback of various table operations sync to Metadata Table correctly.
*/
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
index 8a9f855..58c25a1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
@@ -245,7 +245,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// Metadata is in sync till the latest completed instant on the dataset
HoodieTimer timer = new HoodieTimer().startTimer();
- List<FileSlice> latestFileSlices = HoodieTableMetadataUtil.loadPartitionFileGroupsWithLatestFileSlices(metadataMetaClient, partitionName);
+ List<FileSlice> latestFileSlices = HoodieTableMetadataUtil.loadPartitionFileGroupsWithLatestFileSlices(metadataMetaClient, partitionName, true);
if (latestFileSlices.size() == 0) {
// empty partition
return Pair.of(null, null);
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 9078bd0..b4dfbbd 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -50,6 +50,7 @@ import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.hudi.metadata.HoodieTableMetadata.NON_PARTITIONED_NAME;
@@ -338,9 +339,10 @@ public class HoodieTableMetadataUtil {
* The list of file slices returned is sorted in the correct order of file group name.
* @param metaClient instance of {@link HoodieTableMetaClient}.
* @param partition The name of the partition whose file groups are to be loaded.
+ * @param isReader true if reader code path, false otherwise.
* @return List of latest file slices for all file groups in a given partition.
*/
- public static List<FileSlice> loadPartitionFileGroupsWithLatestFileSlices(HoodieTableMetaClient metaClient, String partition) {
+ public static List<FileSlice> loadPartitionFileGroupsWithLatestFileSlices(HoodieTableMetaClient metaClient, String partition, boolean isReader) {
LOG.info("Loading file groups for metadata table partition " + partition);
// If there are no commits on the metadata table then the table's default FileSystemView will not return any file
@@ -352,7 +354,9 @@ public class HoodieTableMetadataUtil {
}
HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient, timeline);
- return fsView.getLatestFileSlices(partition).sorted((s1, s2) -> s1.getFileId().compareTo(s2.getFileId()))
+ Stream<FileSlice> fileSliceStream = isReader ? fsView.getLatestMergedFileSlicesBeforeOrOn(partition, timeline.filterCompletedInstants().lastInstant().get().getTimestamp()) :
+ fsView.getLatestFileSlices(partition);
+ return fileSliceStream.sorted((s1, s2) -> s1.getFileId().compareTo(s2.getFileId()))
.collect(Collectors.toList());
}
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
index 15215b8..486c473 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
@@ -344,6 +344,16 @@ public class FileCreateUtils {
removeMetaFile(basePath, instantTime, HoodieTimeline.ROLLBACK_EXTENSION);
}
+ public static java.nio.file.Path renameFileToTemp(java.nio.file.Path sourcePath, String instantTime) throws IOException {
+ java.nio.file.Path dummyFilePath = sourcePath.getParent().resolve(instantTime + ".temp");
+ Files.move(sourcePath, dummyFilePath);
+ return dummyFilePath;
+ }
+
+ public static void renameTempToMetaFile(java.nio.file.Path tempFilePath, java.nio.file.Path destPath) throws IOException {
+ Files.move(tempFilePath, destPath);
+ }
+
public static long getTotalMarkerFileCount(String basePath, String partitionPath, String instantTime, IOType ioType) throws IOException {
Path parentPath = Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME, instantTime, partitionPath);
if (Files.notExists(parentPath)) {
[hudi] 10/26: [HUDI-2904] Fix metadata table archival overstepping between regular writers and table services (#4186)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 318818be44d065f1be523f7aa9eb949e09859cbb
Author: rmahindra123 <76...@users.noreply.github.com>
AuthorDate: Thu Dec 2 10:32:26 2021 -0800
[HUDI-2904] Fix metadata table archival overstepping between regular writers and table services (#4186)
- Co-authored-by: Rajesh Mahindra <rm...@Rajeshs-MacBook-Pro.local>
- Co-authored-by: Sivabalan Narayanan <n....@gmail.com>
(cherry picked from commit 91d2e61433e74abb44cb4d0ae236ee8f4a94e1f8)
---
.../hudi/client/AbstractHoodieWriteClient.java | 33 ++++++++++++++++---
.../apache/hudi/config/HoodieCompactionConfig.java | 12 +++++++
.../org/apache/hudi/config/HoodieWriteConfig.java | 4 +++
.../metadata/HoodieBackedTableMetadataWriter.java | 4 ++-
.../apache/hudi/client/HoodieFlinkWriteClient.java | 11 +++----
.../FlinkHoodieBackedTableMetadataWriter.java | 1 +
.../SparkHoodieBackedTableMetadataWriter.java | 1 +
.../functional/TestHoodieBackedMetadata.java | 38 ++++++++++++++++++++++
8 files changed, 91 insertions(+), 13 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
index 96d89fc..59acbb2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
@@ -449,11 +449,9 @@ public abstract class AbstractHoodieWriteClient<T extends HoodieRecordPayload, I
WriteMarkersFactory.get(config.getMarkersType(), table, instantTime)
.quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
autoCleanOnCommit();
- // We cannot have unbounded commit files. Archive commits if we have to archive
- HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, table);
- archiveLog.archiveIfRequired(context);
- } catch (IOException ioe) {
- throw new HoodieIOException(ioe.getMessage(), ioe);
+ if (config.isAutoArchive()) {
+ archive(table);
+ }
} finally {
this.heartbeatClient.stop(instantTime);
}
@@ -744,6 +742,31 @@ public abstract class AbstractHoodieWriteClient<T extends HoodieRecordPayload, I
}
/**
+ * Trigger archival for the table. This ensures that the number of commits do not explode
+ * and keep increasing unbounded over time.
+ * @param table table to commit on.
+ */
+ protected void archive(HoodieTable<T, I, K, O> table) {
+ try {
+ // We cannot have unbounded commit files. Archive commits if we have to archive
+ HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, table);
+ archiveLog.archiveIfRequired(context);
+ } catch (IOException ioe) {
+ throw new HoodieIOException("Failed to archive", ioe);
+ }
+ }
+
+ /**
+ * Trigger archival for the table. This ensures that the number of commits do not explode
+ * and keep increasing unbounded over time.
+ */
+ public void archive() {
+ // Create a Hoodie table which encapsulated the commits and files visible
+ HoodieTable table = createTable(config, hadoopConf);
+ archive(table);
+ }
+
+ /**
* Provides a new commit time for a write operation (insert/update/delete).
*/
public String startCommit() {
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 fbe31b0..640f0cb 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<String> 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<String> ASYNC_CLEAN = ConfigProperty
.key("hoodie.clean.async")
.defaultValue("false")
@@ -493,6 +500,11 @@ public class HoodieCompactionConfig extends HoodieConfig {
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;
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 b49108f..df4b3f6 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
@@ -1101,6 +1101,10 @@ public class HoodieWriteConfig extends HoodieConfig {
return getBoolean(HoodieCompactionConfig.AUTO_CLEAN);
}
+ public boolean isAutoArchive() {
+ return getBoolean(HoodieCompactionConfig.AUTO_ARCHIVE);
+ }
+
public boolean isAsyncClean() {
return getBoolean(HoodieCompactionConfig.ASYNC_CLEAN);
}
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 f9486b1..54284fc 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
@@ -204,7 +204,9 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
.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)
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
index 36caa1b..374dd12 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -40,7 +40,6 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieCommitException;
import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.index.FlinkHoodieIndexFactory;
import org.apache.hudi.index.HoodieIndex;
@@ -57,7 +56,6 @@ import org.apache.hudi.metadata.HoodieBackedTableMetadataWriter;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.table.HoodieFlinkTable;
import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.table.HoodieTimelineArchiveLog;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.hudi.table.action.compact.CompactHelpers;
import org.apache.hudi.table.marker.WriteMarkersFactory;
@@ -332,11 +330,10 @@ public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
// Delete the marker directory for the instant.
WriteMarkersFactory.get(config.getMarkersType(), createTable(config, hadoopConf), instantTime)
.quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
- // We cannot have unbounded commit files. Archive commits if we have to archive
- HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, table);
- archiveLog.archiveIfRequired(context);
- } catch (IOException ioe) {
- throw new HoodieIOException(ioe.getMessage(), ioe);
+ if (config.isAutoArchive()) {
+ // We cannot have unbounded commit files. Archive commits if we have to archive
+ archive(table);
+ }
} finally {
this.heartbeatClient.stop(instantTime);
}
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
index 5e782c5..0dcfcfc 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
@@ -140,6 +140,7 @@ public class FlinkHoodieBackedTableMetadataWriter extends HoodieBackedTableMetad
if (canTriggerTableService) {
compactIfNecessary(writeClient, instantTime);
doClean(writeClient, instantTime);
+ writeClient.archive();
}
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
index ff8f556..65ade82 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
@@ -155,6 +155,7 @@ public class SparkHoodieBackedTableMetadataWriter extends HoodieBackedTableMetad
if (canTriggerTableService) {
compactIfNecessary(writeClient, instantTime);
doClean(writeClient, instantTime);
+ writeClient.archive();
}
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
index 82bc892..73b7811 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
@@ -107,6 +107,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import static java.util.Arrays.asList;
@@ -250,6 +251,43 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
validateMetadata(testTable, emptyList(), true);
}
+ @Test
+ public void testMetadataTableArchival() throws Exception {
+ init(COPY_ON_WRITE, false);
+ writeConfig = getWriteConfigBuilder(true, true, false)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .enableFullScan(true)
+ .enableMetrics(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(3)
+ .archiveCommitsWith(3, 4)
+ .retainCommits(1)
+ .build())
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder().archiveCommitsWith(2, 3).retainCommits(1).build()).build();
+ initWriteConfigAndMetatableWriter(writeConfig, true);
+
+ AtomicInteger commitTime = new AtomicInteger(1);
+ // trigger 2 regular writes(1 bootstrap commit). just 1 before archival can get triggered.
+ int i = 1;
+ for (; i <= 2; i++) {
+ doWriteOperation(testTable, "000000" + (commitTime.getAndIncrement()), INSERT);
+ }
+ // expected num commits = 1 (bootstrap) + 2 (writes) + 1 compaction.
+ HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(metadataTableBasePath).build();
+ HoodieActiveTimeline metadataTimeline = metadataMetaClient.reloadActiveTimeline();
+ assertEquals(metadataTimeline.getCommitsTimeline().filterCompletedInstants().countInstants(), 4);
+
+ // trigger a async table service, archival should not kick in, even though conditions are met.
+ doCluster(testTable, "000000" + commitTime.getAndIncrement());
+ metadataTimeline = metadataMetaClient.reloadActiveTimeline();
+ assertEquals(metadataTimeline.getCommitsTimeline().filterCompletedInstants().countInstants(), 5);
+
+ // trigger a regular write operation. archival should kick in.
+ doWriteOperation(testTable, "000000" + (commitTime.getAndIncrement()), INSERT);
+ metadataTimeline = metadataMetaClient.reloadActiveTimeline();
+ assertEquals(metadataTimeline.getCommitsTimeline().filterCompletedInstants().countInstants(), 3);
+ }
+
@ParameterizedTest
@EnumSource(HoodieTableType.class)
public void testMetadataInsertUpsertClean(HoodieTableType tableType) throws Exception {
[hudi] 05/26: Revert "[HUDI-2856] Bit cask disk map delete modified (#4116)" (#4171)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 7df81a6b6ff5405e9da93289789ec84a280cf8d7
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Tue Nov 30 22:08:44 2021 -0800
Revert "[HUDI-2856] Bit cask disk map delete modified (#4116)" (#4171)
This reverts commit 257a6a74569cdb500b06c53b92d05f6f5833aa31.
(cherry picked from commit 9b254b6fc5779b6105b2fa9aeff5ad5b79407f84)
---
.../java/org/apache/hudi/common/util/collection/BitCaskDiskMap.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/BitCaskDiskMap.java b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/BitCaskDiskMap.java
index 7644afa..289901d 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/BitCaskDiskMap.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/BitCaskDiskMap.java
@@ -281,12 +281,12 @@ public final class BitCaskDiskMap<T extends Serializable, R extends Serializable
}
}
}
+ writeOnlyFile.delete();
+ this.iterators.forEach(ClosableIterator::close);
} catch (Exception e) {
// delete the file for any sort of exception
- LOG.error("BitCaskDisMap close error ", e);
- } finally {
- this.iterators.forEach(ClosableIterator::close);
writeOnlyFile.delete();
+ } finally {
super.close();
}
}
[hudi] 22/26: [HUDI-2890] Kafka Connect: Fix failed writes and avoid table service concurrent operations (#4211)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit b88d4beb23a5ba48fbe642d1f83e55a332b7ce5d
Author: rmahindra123 <76...@users.noreply.github.com>
AuthorDate: Fri Dec 3 21:30:32 2021 -0800
[HUDI-2890] Kafka Connect: Fix failed writes and avoid table service concurrent operations (#4211)
* Fix kafka connect readme
* Fix handling of errors in write records for kafka connect
* By default, ensure we skip error records and keep the pipeline alive
* Fix indentation
Co-authored-by: Rajesh Mahindra <rm...@Rajeshs-MacBook-Pro.local>
(cherry picked from commit 94f45e928c917c87934badb2bb4bc8c767a14e3a)
---
hudi-kafka-connect/README.md | 71 +++++++------
hudi-kafka-connect/demo/setupKafka.sh | 2 +-
.../transaction/ConnectTransactionCoordinator.java | 55 +++++++---
.../writers/ConnectTransactionServices.java | 2 +-
.../hudi/connect/writers/KafkaConnectConfigs.java | 14 +++
.../writers/KafkaConnectTransactionServices.java | 30 +++---
.../writers/KafkaConnectWriterProvider.java | 12 ++-
.../connect/TestConnectTransactionCoordinator.java | 117 +++++++++++++++------
.../helper/MockConnectTransactionServices.java | 3 +-
.../test/resources/log4j-surefire-quiet.properties | 30 ++++++
.../src/test/resources/log4j-surefire.properties | 31 ++++++
11 files changed, 271 insertions(+), 96 deletions(-)
diff --git a/hudi-kafka-connect/README.md b/hudi-kafka-connect/README.md
index 0bf53d6..9d63f5a 100644
--- a/hudi-kafka-connect/README.md
+++ b/hudi-kafka-connect/README.md
@@ -54,35 +54,17 @@ mvn package -DskipTests -pl packaging/hudi-kafka-connect-bundle -am
cp $HUDI_DIR/packaging/hudi-kafka-connect-bundle/target/hudi-kafka-connect-bundle-0.10.0-SNAPSHOT.jar /usr/local/share/kafka/plugins/lib
```
-### 2 - Set up the docker containers
-
-To run the connect locally, we need kafka, zookeeper, hdfs, hive etc. To make the setup easier, we use the docker
-containers from the hudi docker demo. Refer to [this link for the setup](https://hudi.apache.org/docs/docker_demo)
-
-Essentially, follow the steps listed here:
-
-/etc/hosts : The demo references many services running in container by the hostname. Add the following settings to /etc/hosts
-```bash
-127.0.0.1 adhoc-1
-127.0.0.1 adhoc-2
-127.0.0.1 namenode
-127.0.0.1 datanode1
-127.0.0.1 hiveserver
-127.0.0.1 hivemetastore
-127.0.0.1 kafkabroker
-127.0.0.1 sparkmaster
-127.0.0.1 zookeeper
-```
-
-Bring up the docker containers
+Set up a Kafka broker locally. Download the latest apache kafka from [here](https://kafka.apache.org/downloads).
+Once downloaded and built, run the Zookeeper server and Kafka server using the command line tools.
```bash
-cd ${HUDI_DIR}/docker
-./setup_demo.sh
+export KAFKA_HOME=/path/to/kafka_install_dir
+cd $KAFKA_HOME
+./bin/zookeeper-server-start.sh ./config/zookeeper.properties
+./bin/kafka-server-start.sh ./config/server.properties
```
+Wait until the kafka cluster is up and running.
-The schema registry and kafka connector can be run from host system directly (mac/ linux).
-
-### 3 - Set up the schema registry
+### 2 - Set up the schema registry
Hudi leverages schema registry to obtain the latest schema when writing records. While it supports most popular schema
registries, we use Confluent schema registry. Download the
@@ -97,7 +79,7 @@ cd $CONFLUENT_DIR
./bin/schema-registry-start etc/schema-registry/schema-registry.properties
```
-### 4 - Create the Hudi Control Topic for Coordination of the transactions
+### 3 - Create the Hudi Control Topic for Coordination of the transactions
The control topic should only have `1` partition, since its used to coordinate the Hudi write transactions across the multiple Connect tasks.
@@ -107,7 +89,7 @@ cd $KAFKA_HOME
./bin/kafka-topics.sh --create --topic hudi-control-topic --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092
```
-### 5 - Create the Hudi Topic for the Sink and insert data into the topic
+### 4 - Create the Hudi Topic for the Sink and insert data into the topic
Open a terminal to execute the following command:
@@ -123,7 +105,7 @@ to generate, with each batch containing a number of messages and idle time betwe
bash setupKafka.sh -n <num_kafka_messages_per_batch> -b <num_batches>
```
-### 6 - Run the Sink connector worker (multiple workers can be run)
+### 5 - Run the Sink connector worker (multiple workers can be run)
The Kafka connect is a distributed platform, with the ability to run one or more workers (each running multiple tasks)
that parallely process the records from the Kafka partitions for the same topic. We provide a properties file with
@@ -137,7 +119,7 @@ cd ${KAFKA_HOME}
./bin/connect-distributed.sh ${HUDI_DIR}/hudi-kafka-connect/demo/connect-distributed.properties
```
-### 7 - To add the Hudi Sink to the Connector (delete it if you want to re-configure)
+### 6 - To add the Hudi Sink to the Connector (delete it if you want to re-configure)
Once the Connector has started, it will not run the Sink, until the Hudi sink is added using the web api. The following
curl APIs can be used to delete and add a new Hudi Sink. Again, a default configuration is provided for the Hudi Sink,
@@ -188,7 +170,7 @@ total 5168
-rw-r--r-- 1 user wheel 440214 Sep 13 21:43 E200FA75DCD1CED60BE86BCE6BF5D23A-0_0-0-0_20210913214114.parquet
```
-### 8 - Run async compaction and clustering if scheduled
+### 7 - Run async compaction and clustering if scheduled
When using Merge-On-Read (MOR) as the table type, async compaction and clustering can be scheduled when the Sink is
running. Inline compaction and clustering are disabled by default due to performance reason. By default, async
@@ -318,11 +300,36 @@ hoodie.write.concurrency.mode=single_writer
Note that you don't have to provide the instant time through `--instant-time`. In that case, the earliest scheduled
clustering is going to be executed.
-### 9- Querying via Hive
+### 8- Querying via Hive
In this section we explain how one can test syncing of the Hudi table with Hive server/ Hive Metastore,
that enable querying via Hive, Presto etc.
+To ease the deployment of HDFS, Hive Server, Hive Metastore etc. for testing hive sync, we use the docker
+containers from the hudi docker demo. Refer to [this link for the setup](https://hudi.apache.org/docs/docker_demo).
+Additionally, the docker deploys kafka and zookeeper too, so you do not need to run them explicitly in this setup.
+
+Essentially, follow the steps listed here:
+
+/etc/hosts : The demo references many services running in container by the hostname. Add the following settings to /etc/hosts
+```bash
+127.0.0.1 adhoc-1
+127.0.0.1 adhoc-2
+127.0.0.1 namenode
+127.0.0.1 datanode1
+127.0.0.1 hiveserver
+127.0.0.1 hivemetastore
+127.0.0.1 kafkabroker
+127.0.0.1 sparkmaster
+127.0.0.1 zookeeper
+```
+
+Bring up the docker containers
+```bash
+cd ${HUDI_DIR}/docker
+./setup_demo.sh
+```
+
Firstly, (re)-install a different connector that is configured to write the Hudi table to Hdfs instead of local filesystem.
```bash
diff --git a/hudi-kafka-connect/demo/setupKafka.sh b/hudi-kafka-connect/demo/setupKafka.sh
index bc615a5..4915564 100755
--- a/hudi-kafka-connect/demo/setupKafka.sh
+++ b/hudi-kafka-connect/demo/setupKafka.sh
@@ -50,7 +50,7 @@ fi
## defaults
rawDataFile=${HUDI_DIR}/docker/demo/data/batch_1.json
-kafkaBrokerHostname=kafkabroker
+kafkaBrokerHostname=localhost
kafkaTopicName=hudi-test-topic
numKafkaPartitions=4
recordKey=volume
diff --git a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java
index 7acd875..14fd880 100644
--- a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java
+++ b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java
@@ -280,26 +280,56 @@ public class ConnectTransactionCoordinator implements TransactionCoordinator, Ru
private void onReceiveWriteStatus(ControlMessage message) {
ControlMessage.ParticipantInfo participantInfo = message.getParticipantInfo();
- int partition = message.getSenderPartition();
- partitionsWriteStatusReceived.put(partition, KafkaConnectUtils.getWriteStatuses(participantInfo));
- currentConsumedKafkaOffsets.put(partition, participantInfo.getKafkaOffset());
+ int partitionId = message.getSenderPartition();
+ partitionsWriteStatusReceived.put(partitionId, KafkaConnectUtils.getWriteStatuses(participantInfo));
+ currentConsumedKafkaOffsets.put(partitionId, participantInfo.getKafkaOffset());
if (partitionsWriteStatusReceived.size() >= numPartitions
&& currentState.equals(State.ENDED_COMMIT)) {
// Commit the kafka offsets to the commit file
try {
List<WriteStatus> allWriteStatuses = new ArrayList<>();
partitionsWriteStatusReceived.forEach((key, value) -> allWriteStatuses.addAll(value));
- // Commit the last write in Hudi, along with the latest kafka offset
- if (!allWriteStatuses.isEmpty()) {
- transactionServices.endCommit(currentCommitTime,
+
+ long totalErrorRecords = (long) allWriteStatuses.stream().mapToDouble(WriteStatus::getTotalErrorRecords).sum();
+ long totalRecords = (long) allWriteStatuses.stream().mapToDouble(WriteStatus::getTotalRecords).sum();
+ boolean hasErrors = totalErrorRecords > 0;
+
+ if ((!hasErrors || configs.allowCommitOnErrors()) && !allWriteStatuses.isEmpty()) {
+ boolean success = transactionServices.endCommit(currentCommitTime,
allWriteStatuses,
transformKafkaOffsets(currentConsumedKafkaOffsets));
+
+ if (success) {
+ LOG.info("Commit " + currentCommitTime + " successful!");
+ currentState = State.WRITE_STATUS_RCVD;
+ globalCommittedKafkaOffsets.putAll(currentConsumedKafkaOffsets);
+ submitEvent(new CoordinatorEvent(CoordinatorEvent.CoordinatorEventType.ACK_COMMIT,
+ message.getTopicName(),
+ currentCommitTime));
+ return;
+ } else {
+ LOG.error("Commit " + currentCommitTime + " failed!");
+ }
+ } else if (hasErrors) {
+ LOG.error("Coordinator found errors when writing. Errors/Total=" + totalErrorRecords + "/" + totalRecords);
+ LOG.error("Printing out the top 100 errors");
+ allWriteStatuses.stream().filter(WriteStatus::hasErrors).limit(100).forEach(ws -> {
+ LOG.error("Global error :", ws.getGlobalError());
+ if (ws.getErrors().size() > 0) {
+ ws.getErrors().forEach((key, value) -> LOG.trace("Error for key:" + key + " is " + value));
+ }
+ });
+ } else {
+ LOG.warn("Empty write statuses were received from all Participants");
}
- currentState = State.WRITE_STATUS_RCVD;
- globalCommittedKafkaOffsets.putAll(currentConsumedKafkaOffsets);
- submitEvent(new CoordinatorEvent(CoordinatorEvent.CoordinatorEventType.ACK_COMMIT,
- message.getTopicName(),
- currentCommitTime));
+
+ // Submit the next start commit, that will rollback the current commit.
+ currentState = State.FAILED_COMMIT;
+ LOG.warn("Current commit " + currentCommitTime + " failed, so starting a new commit after recovery delay");
+ submitEvent(new CoordinatorEvent(CoordinatorEvent.CoordinatorEventType.START_COMMIT,
+ partition.topic(),
+ StringUtils.EMPTY_STRING),
+ RESTART_COMMIT_DELAY_MS, TimeUnit.MILLISECONDS);
} catch (Exception exception) {
LOG.error("Fatal error while committing file", exception);
}
@@ -310,7 +340,7 @@ public class ConnectTransactionCoordinator implements TransactionCoordinator, Ru
// If we are still stuck in ENDED_STATE
if (currentState.equals(State.ENDED_COMMIT)) {
currentState = State.WRITE_STATUS_TIMEDOUT;
- LOG.warn("Did not receive the Write Status from all partitions");
+ LOG.warn("Current commit " + currentCommitTime + " failed after a write status timeout, so starting a new commit after recovery delay");
// Submit the next start commit
submitEvent(new CoordinatorEvent(CoordinatorEvent.CoordinatorEventType.START_COMMIT,
partition.topic(),
@@ -365,6 +395,7 @@ public class ConnectTransactionCoordinator implements TransactionCoordinator, Ru
INIT,
STARTED_COMMIT,
ENDED_COMMIT,
+ FAILED_COMMIT,
WRITE_STATUS_RCVD,
WRITE_STATUS_TIMEDOUT,
ACKED_COMMIT,
diff --git a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java
index b36e1f1..2ce44ff 100644
--- a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java
+++ b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java
@@ -32,7 +32,7 @@ public interface ConnectTransactionServices {
String startCommit();
- void endCommit(String commitTime, List<WriteStatus> writeStatuses, Map<String, String> extraMetadata);
+ boolean endCommit(String commitTime, List<WriteStatus> writeStatuses, Map<String, String> extraMetadata);
Map<String, String> fetchLatestExtraCommitMetadata();
}
diff --git a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectConfigs.java b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectConfigs.java
index 692edf7..ec03451 100644
--- a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectConfigs.java
+++ b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectConfigs.java
@@ -88,6 +88,11 @@ public class KafkaConnectConfigs extends HoodieConfig {
.defaultValue(HiveSyncTool.class.getName())
.withDocumentation("Meta sync client tool, using comma to separate multi tools");
+ public static final ConfigProperty<Boolean> ALLOW_COMMIT_ON_ERRORS = ConfigProperty
+ .key("hoodie.kafka.allow.commit.on.errors")
+ .defaultValue(true)
+ .withDocumentation("Commit even when some records failed to be written");
+
protected KafkaConnectConfigs() {
super();
}
@@ -136,6 +141,10 @@ public class KafkaConnectConfigs extends HoodieConfig {
return getString(META_SYNC_CLASSES);
}
+ public Boolean allowCommitOnErrors() {
+ return getBoolean(ALLOW_COMMIT_ON_ERRORS);
+ }
+
public static class Builder {
protected final KafkaConnectConfigs connectConfigs = new KafkaConnectConfigs();
@@ -160,6 +169,11 @@ public class KafkaConnectConfigs extends HoodieConfig {
return this;
}
+ public Builder withAllowCommitOnErrors(Boolean allowCommitOnErrors) {
+ connectConfigs.setValue(ALLOW_COMMIT_ON_ERRORS, String.valueOf(allowCommitOnErrors));
+ return this;
+ }
+
// Kafka connect task are passed with props with type Map<>
public Builder withProperties(Map<?, ?> properties) {
connectConfigs.getProps().putAll(properties);
diff --git a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
index b18e7c6..cca738a 100644
--- a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
+++ b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
@@ -77,6 +77,7 @@ public class KafkaConnectTransactionServices implements ConnectTransactionServic
public KafkaConnectTransactionServices(KafkaConnectConfigs connectConfigs) throws HoodieException {
this.connectConfigs = connectConfigs;
+ // This is the writeConfig for the Transaction Coordinator
this.writeConfig = HoodieWriteConfig.newBuilder()
.withEngineType(EngineType.JAVA)
.withProperties(connectConfigs.getProps())
@@ -122,20 +123,23 @@ public class KafkaConnectTransactionServices implements ConnectTransactionServic
}
@Override
- public void endCommit(String commitTime, List<WriteStatus> writeStatuses, Map<String, String> extraMetadata) {
- javaClient.commit(commitTime, writeStatuses, Option.of(extraMetadata));
- LOG.info("Ending Hudi commit " + commitTime);
-
- // Schedule clustering and compaction as needed.
- if (writeConfig.isAsyncClusteringEnabled()) {
- javaClient.scheduleClustering(Option.empty()).ifPresent(
- instantTs -> LOG.info("Scheduled clustering at instant time:" + instantTs));
- }
- if (isAsyncCompactionEnabled()) {
- javaClient.scheduleCompaction(Option.empty()).ifPresent(
- instantTs -> LOG.info("Scheduled compaction at instant time:" + instantTs));
+ public boolean endCommit(String commitTime, List<WriteStatus> writeStatuses, Map<String, String> extraMetadata) {
+ boolean success = javaClient.commit(commitTime, writeStatuses, Option.of(extraMetadata));
+ if (success) {
+ LOG.info("Ending Hudi commit " + commitTime);
+
+ // Schedule clustering and compaction as needed.
+ if (writeConfig.isAsyncClusteringEnabled()) {
+ javaClient.scheduleClustering(Option.empty()).ifPresent(
+ instantTs -> LOG.info("Scheduled clustering at instant time:" + instantTs));
+ }
+ if (isAsyncCompactionEnabled()) {
+ javaClient.scheduleCompaction(Option.empty()).ifPresent(
+ instantTs -> LOG.info("Scheduled compaction at instant time:" + instantTs));
+ }
+ syncMeta();
}
- syncMeta();
+ return success;
}
@Override
diff --git a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
index 5bac1f0..6ab0469 100644
--- a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
+++ b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
@@ -27,6 +27,8 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieAvroPayload;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.connect.KafkaConnectFileIdPrefixProvider;
@@ -73,7 +75,7 @@ public class KafkaConnectWriterProvider implements ConnectWriterProvider<WriteSt
this.keyGenerator = HoodieAvroKeyGeneratorFactory.createKeyGenerator(
new TypedProperties(connectConfigs.getProps()));
- // Create the write client to write some records in
+ // This is the writeConfig for the writers for the individual Transaction Coordinators
writeConfig = HoodieWriteConfig.newBuilder()
.withEngineType(EngineType.JAVA)
.withProperties(connectConfigs.getProps())
@@ -84,6 +86,14 @@ public class KafkaConnectWriterProvider implements ConnectWriterProvider<WriteSt
.withSchema(schemaProvider.getSourceSchema().toString())
.withAutoCommit(false)
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ // participants should not trigger table services, and leave it to the coordinator
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withAutoClean(false)
+ .withAutoArchive(false)
+ .withInlineCompaction(false).build())
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withInlineClustering(false)
+ .build())
.build();
context = new HoodieJavaEngineContext(hadoopConf);
diff --git a/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java b/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java
index 6e049c6..f003fe9 100644
--- a/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java
+++ b/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java
@@ -50,7 +50,7 @@ import static org.mockito.Mockito.mock;
public class TestConnectTransactionCoordinator {
private static final String TOPIC_NAME = "kafka-connect-test-topic";
- private static final int NUM_PARTITIONS = 4;
+ private static final int TOTAL_KAFKA_PARTITIONS = 4;
private static final int MAX_COMMIT_ROUNDS = 5;
private static final int TEST_TIMEOUT_SECS = 60;
@@ -63,10 +63,6 @@ public class TestConnectTransactionCoordinator {
@BeforeEach
public void setUp() throws Exception {
transactionServices = new MockConnectTransactionServices();
- configs = KafkaConnectConfigs.newBuilder()
- .withCommitIntervalSecs(1L)
- .withCoordinatorWriteTimeoutSecs(1L)
- .build();
latch = new CountDownLatch(1);
}
@@ -77,13 +73,22 @@ public class TestConnectTransactionCoordinator {
participant = new MockParticipant(kafkaControlAgent, latch, scenario, MAX_COMMIT_ROUNDS);
participant.start();
+ KafkaConnectConfigs.Builder configBuilder = KafkaConnectConfigs.newBuilder()
+ .withCommitIntervalSecs(1L)
+ .withCoordinatorWriteTimeoutSecs(1L);
+
+ if (scenario.equals(MockParticipant.TestScenarios.SUBSET_WRITE_STATUS_FAILED)) {
+ configBuilder.withAllowCommitOnErrors(false);
+ }
+ configs = configBuilder.build();
+
// Test the coordinator using the mock participant
TransactionCoordinator coordinator = new ConnectTransactionCoordinator(
configs,
new TopicPartition(TOPIC_NAME, 0),
kafkaControlAgent,
transactionServices,
- (bootstrapServers, topicName) -> NUM_PARTITIONS);
+ (bootstrapServers, topicName) -> TOTAL_KAFKA_PARTITIONS);
coordinator.start();
latch.await(TEST_TIMEOUT_SECS, TimeUnit.SECONDS);
@@ -119,7 +124,7 @@ public class TestConnectTransactionCoordinator {
this.latch = latch;
this.testScenario = testScenario;
this.maxNumberCommitRounds = maxNumberCommitRounds;
- this.partition = new TopicPartition(TOPIC_NAME, (NUM_PARTITIONS - 1));
+ this.partition = new TopicPartition(TOPIC_NAME, (TOTAL_KAFKA_PARTITIONS - 1));
this.kafkaOffsetsCommitted = new HashMap<>();
expectedMsgType = ControlMessage.EventType.START_COMMIT;
numberCommitRounds = 0;
@@ -162,39 +167,40 @@ public class TestConnectTransactionCoordinator {
private void testScenarios(ControlMessage message) {
assertEquals(expectedMsgType, message.getType());
-
switch (message.getType()) {
case START_COMMIT:
expectedMsgType = ControlMessage.EventType.END_COMMIT;
break;
case END_COMMIT:
assertEquals(kafkaOffsetsCommitted, message.getCoordinatorInfo().getGlobalKafkaCommitOffsets());
- int numSuccessPartitions;
+ int numPartitionsThatReportWriteStatus;
Map<Integer, Long> kafkaOffsets = new HashMap<>();
List<ControlMessage> controlEvents = new ArrayList<>();
- // Prepare the WriteStatuses for all partitions
- for (int i = 1; i <= NUM_PARTITIONS; i++) {
- try {
- long kafkaOffset = (long) (Math.random() * 10000);
- kafkaOffsets.put(i, kafkaOffset);
- ControlMessage event = successWriteStatus(
- message.getCommitTime(),
- new TopicPartition(TOPIC_NAME, i),
- kafkaOffset);
- controlEvents.add(event);
- } catch (Exception exception) {
- throw new HoodieException("Fatal error sending control event to Coordinator");
- }
- }
-
switch (testScenario) {
case ALL_CONNECT_TASKS_SUCCESS:
- numSuccessPartitions = NUM_PARTITIONS;
+ composeControlEvent(message.getCommitTime(), false, kafkaOffsets, controlEvents);
+ numPartitionsThatReportWriteStatus = TOTAL_KAFKA_PARTITIONS;
+ // This commit round should succeed, and the kafka offsets getting committed
+ kafkaOffsetsCommitted.putAll(kafkaOffsets);
+ expectedMsgType = ControlMessage.EventType.ACK_COMMIT;
+ break;
+ case SUBSET_WRITE_STATUS_FAILED_BUT_IGNORED:
+ composeControlEvent(message.getCommitTime(), true, kafkaOffsets, controlEvents);
+ numPartitionsThatReportWriteStatus = TOTAL_KAFKA_PARTITIONS;
+ // Despite error records, this commit round should succeed, and the kafka offsets getting committed
kafkaOffsetsCommitted.putAll(kafkaOffsets);
expectedMsgType = ControlMessage.EventType.ACK_COMMIT;
break;
+ case SUBSET_WRITE_STATUS_FAILED:
+ composeControlEvent(message.getCommitTime(), true, kafkaOffsets, controlEvents);
+ numPartitionsThatReportWriteStatus = TOTAL_KAFKA_PARTITIONS;
+ // This commit round should fail, and a new commit round should start without kafka offsets getting committed
+ expectedMsgType = ControlMessage.EventType.START_COMMIT;
+ break;
case SUBSET_CONNECT_TASKS_FAILED:
- numSuccessPartitions = NUM_PARTITIONS / 2;
+ composeControlEvent(message.getCommitTime(), false, kafkaOffsets, controlEvents);
+ numPartitionsThatReportWriteStatus = TOTAL_KAFKA_PARTITIONS / 2;
+ // This commit round should fail, and a new commit round should start without kafka offsets getting committed
expectedMsgType = ControlMessage.EventType.START_COMMIT;
break;
default:
@@ -202,7 +208,7 @@ public class TestConnectTransactionCoordinator {
}
// Send events based on test scenario
- for (int i = 0; i < numSuccessPartitions; i++) {
+ for (int i = 0; i < numPartitionsThatReportWriteStatus; i++) {
kafkaControlAgent.publishMessage(controlEvents.get(i));
}
break;
@@ -227,18 +233,36 @@ public class TestConnectTransactionCoordinator {
public enum TestScenarios {
SUBSET_CONNECT_TASKS_FAILED,
+ SUBSET_WRITE_STATUS_FAILED,
+ SUBSET_WRITE_STATUS_FAILED_BUT_IGNORED,
ALL_CONNECT_TASKS_SUCCESS
}
- private static ControlMessage successWriteStatus(String commitTime,
- TopicPartition partition,
- long kafkaOffset) throws Exception {
- // send WS
- WriteStatus writeStatus = new WriteStatus();
- WriteStatus status = new WriteStatus(false, 1.0);
- for (int i = 0; i < 1000; i++) {
- status.markSuccess(mock(HoodieRecord.class), Option.empty());
+ private static void composeControlEvent(String commitTime, boolean shouldIncludeFailedRecords, Map<Integer, Long> kafkaOffsets, List<ControlMessage> controlEvents) {
+ // Prepare the WriteStatuses for all partitions
+ for (int i = 1; i <= TOTAL_KAFKA_PARTITIONS; i++) {
+ try {
+ long kafkaOffset = (long) (Math.random() * 10000);
+ kafkaOffsets.put(i, kafkaOffset);
+ ControlMessage event = composeWriteStatusResponse(
+ commitTime,
+ new TopicPartition(TOPIC_NAME, i),
+ kafkaOffset,
+ shouldIncludeFailedRecords);
+ controlEvents.add(event);
+ } catch (Exception exception) {
+ throw new HoodieException("Fatal error sending control event to Coordinator");
+ }
}
+ }
+
+ private static ControlMessage composeWriteStatusResponse(String commitTime,
+ TopicPartition partition,
+ long kafkaOffset,
+ boolean includeFailedRecords) throws Exception {
+ // send WS
+ WriteStatus writeStatus = includeFailedRecords ? getSubsetFailedRecordsWriteStatus() : getAllSuccessfulRecordsWriteStatus();
+
return ControlMessage.newBuilder()
.setType(ControlMessage.EventType.WRITE_STATUS)
.setTopicName(partition.topic())
@@ -255,4 +279,27 @@ public class TestConnectTransactionCoordinator {
).build();
}
}
+
+ private static WriteStatus getAllSuccessfulRecordsWriteStatus() {
+ // send WS
+ WriteStatus status = new WriteStatus(false, 0.0);
+ for (int i = 0; i < 1000; i++) {
+ status.markSuccess(mock(HoodieRecord.class), Option.empty());
+ }
+ return status;
+ }
+
+ private static WriteStatus getSubsetFailedRecordsWriteStatus() {
+ // send WS
+ WriteStatus status = new WriteStatus(false, 0.0);
+ for (int i = 0; i < 1000; i++) {
+ if (i % 10 == 0) {
+ status.markFailure(mock(HoodieRecord.class), new Throwable("Error writing record on disk"), Option.empty());
+ } else {
+ status.markSuccess(mock(HoodieRecord.class), Option.empty());
+ }
+ }
+ status.setGlobalError(new Throwable("More than one records failed to be written to storage"));
+ return status;
+ }
}
diff --git a/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java b/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java
index 6994c65..b3314ad 100644
--- a/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java
+++ b/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java
@@ -46,8 +46,9 @@ public class MockConnectTransactionServices implements ConnectTransactionService
}
@Override
- public void endCommit(String commitTime, List<WriteStatus> writeStatuses, Map<String, String> extraMetadata) {
+ public boolean endCommit(String commitTime, List<WriteStatus> writeStatuses, Map<String, String> extraMetadata) {
assertEquals(String.valueOf(this.commitTime), commitTime);
+ return true;
}
@Override
diff --git a/hudi-kafka-connect/src/test/resources/log4j-surefire-quiet.properties b/hudi-kafka-connect/src/test/resources/log4j-surefire-quiet.properties
new file mode 100644
index 0000000..ca0a50c
--- /dev/null
+++ b/hudi-kafka-connect/src/test/resources/log4j-surefire-quiet.properties
@@ -0,0 +1,30 @@
+###
+# 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.
+###
+log4j.rootLogger=WARN, CONSOLE
+log4j.logger.org.apache.hudi=DEBUG
+log4j.logger.org.apache.hadoop.hbase=ERROR
+
+# CONSOLE is set to be a ConsoleAppender.
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+# CONSOLE uses PatternLayout.
+log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=[%-5p] %d %c %x - %m%n
+log4j.appender.CONSOLE.filter.a=org.apache.log4j.varia.LevelRangeFilter
+log4j.appender.CONSOLE.filter.a.AcceptOnMatch=true
+log4j.appender.CONSOLE.filter.a.LevelMin=WARN
+log4j.appender.CONSOLE.filter.a.LevelMax=FATAL
diff --git a/hudi-kafka-connect/src/test/resources/log4j-surefire.properties b/hudi-kafka-connect/src/test/resources/log4j-surefire.properties
new file mode 100644
index 0000000..32af462
--- /dev/null
+++ b/hudi-kafka-connect/src/test/resources/log4j-surefire.properties
@@ -0,0 +1,31 @@
+###
+# 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.
+###
+log4j.rootLogger=WARN, CONSOLE
+log4j.logger.org.apache=INFO
+log4j.logger.org.apache.hudi=DEBUG
+log4j.logger.org.apache.hadoop.hbase=ERROR
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+# A1 uses PatternLayout.
+log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
+log4j.appender.CONSOLE.filter.a=org.apache.log4j.varia.LevelRangeFilter
+log4j.appender.CONSOLE.filter.a.AcceptOnMatch=true
+log4j.appender.CONSOLE.filter.a.LevelMin=WARN
+log4j.appender.CONSOLE.filter.a.LevelMax=FATAL
[hudi] 21/26: [HUDI-2933] DISABLE Metadata table by default (#4213)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 7034d1a1be7ecefc346aea95a9ff30bdf758204a
Author: vinoth chandar <vi...@users.noreply.github.com>
AuthorDate: Fri Dec 3 21:12:35 2021 -0800
[HUDI-2933] DISABLE Metadata table by default (#4213)
(cherry picked from commit 0fd6b2d71e50a3b09336b74074a2f5117ba57f5c)
---
.../hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java | 2 +-
.../main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
index d359997..5eee262 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
@@ -143,7 +143,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
@ValueSource(booleans = {true, false})
void testRollbackWithDeltaAndCompactionCommit(boolean rollbackUsingMarkers) throws Exception {
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder(false, rollbackUsingMarkers, HoodieIndex.IndexType.SIMPLE)
- .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(true).build());
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build());
addConfigsForPopulateMetaFields(cfgBuilder, true);
HoodieWriteConfig cfg = cfgBuilder.build();
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
index 51791c9..6beb7d1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
@@ -44,7 +44,7 @@ public final class HoodieMetadataConfig extends HoodieConfig {
// Enable the internal Metadata Table which saves file listings
public static final ConfigProperty<Boolean> ENABLE = ConfigProperty
.key(METADATA_PREFIX + ".enable")
- .defaultValue(true)
+ .defaultValue(false)
.sinceVersion("0.7.0")
.withDocumentation("Enable the internal metadata table which serves table metadata like level file listings");
[hudi] 26/26: Bumping release candidate number 3
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e9e693973fd8be77f2039ad1a9ac4b4ddd52124b
Author: yuzhao.cyz <yu...@gmail.com>
AuthorDate: Sat Dec 4 17:39:49 2021 +0800
Bumping release candidate number 3
---
docker/hoodie/hadoop/base/pom.xml | 2 +-
docker/hoodie/hadoop/datanode/pom.xml | 2 +-
docker/hoodie/hadoop/historyserver/pom.xml | 2 +-
docker/hoodie/hadoop/hive_base/pom.xml | 2 +-
docker/hoodie/hadoop/namenode/pom.xml | 2 +-
docker/hoodie/hadoop/pom.xml | 2 +-
docker/hoodie/hadoop/prestobase/pom.xml | 2 +-
docker/hoodie/hadoop/spark_base/pom.xml | 2 +-
docker/hoodie/hadoop/sparkadhoc/pom.xml | 2 +-
docker/hoodie/hadoop/sparkmaster/pom.xml | 2 +-
docker/hoodie/hadoop/sparkworker/pom.xml | 2 +-
hudi-aws/pom.xml | 4 ++--
hudi-cli/pom.xml | 2 +-
hudi-client/hudi-client-common/pom.xml | 4 ++--
hudi-client/hudi-flink-client/pom.xml | 4 ++--
hudi-client/hudi-java-client/pom.xml | 4 ++--
hudi-client/hudi-spark-client/pom.xml | 4 ++--
hudi-client/pom.xml | 2 +-
hudi-common/pom.xml | 2 +-
hudi-examples/pom.xml | 2 +-
hudi-flink/pom.xml | 2 +-
hudi-hadoop-mr/pom.xml | 2 +-
hudi-integ-test/pom.xml | 2 +-
hudi-kafka-connect/pom.xml | 4 ++--
hudi-spark-datasource/hudi-spark-common/pom.xml | 4 ++--
hudi-spark-datasource/hudi-spark/pom.xml | 4 ++--
hudi-spark-datasource/hudi-spark2/pom.xml | 4 ++--
hudi-spark-datasource/hudi-spark3/pom.xml | 4 ++--
hudi-spark-datasource/pom.xml | 2 +-
hudi-sync/hudi-dla-sync/pom.xml | 2 +-
hudi-sync/hudi-hive-sync/pom.xml | 2 +-
hudi-sync/hudi-sync-common/pom.xml | 2 +-
hudi-sync/pom.xml | 2 +-
hudi-timeline-service/pom.xml | 2 +-
hudi-utilities/pom.xml | 2 +-
packaging/hudi-flink-bundle/pom.xml | 2 +-
packaging/hudi-hadoop-mr-bundle/pom.xml | 2 +-
packaging/hudi-hive-sync-bundle/pom.xml | 2 +-
packaging/hudi-integ-test-bundle/pom.xml | 2 +-
packaging/hudi-kafka-connect-bundle/pom.xml | 2 +-
packaging/hudi-presto-bundle/pom.xml | 2 +-
packaging/hudi-spark-bundle/pom.xml | 2 +-
packaging/hudi-timeline-server-bundle/pom.xml | 2 +-
packaging/hudi-utilities-bundle/pom.xml | 2 +-
pom.xml | 2 +-
45 files changed, 55 insertions(+), 55 deletions(-)
diff --git a/docker/hoodie/hadoop/base/pom.xml b/docker/hoodie/hadoop/base/pom.xml
index 86b618a..bb1b44f 100644
--- a/docker/hoodie/hadoop/base/pom.xml
+++ b/docker/hoodie/hadoop/base/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/datanode/pom.xml b/docker/hoodie/hadoop/datanode/pom.xml
index 1476594..5b89b81 100644
--- a/docker/hoodie/hadoop/datanode/pom.xml
+++ b/docker/hoodie/hadoop/datanode/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/historyserver/pom.xml b/docker/hoodie/hadoop/historyserver/pom.xml
index 6df63de..1b7814f 100644
--- a/docker/hoodie/hadoop/historyserver/pom.xml
+++ b/docker/hoodie/hadoop/historyserver/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/hive_base/pom.xml b/docker/hoodie/hadoop/hive_base/pom.xml
index d7d9ccb..a3ac945 100644
--- a/docker/hoodie/hadoop/hive_base/pom.xml
+++ b/docker/hoodie/hadoop/hive_base/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/namenode/pom.xml b/docker/hoodie/hadoop/namenode/pom.xml
index 77218bb..0666eb1 100644
--- a/docker/hoodie/hadoop/namenode/pom.xml
+++ b/docker/hoodie/hadoop/namenode/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/pom.xml b/docker/hoodie/hadoop/pom.xml
index fe316c5..171a096 100644
--- a/docker/hoodie/hadoop/pom.xml
+++ b/docker/hoodie/hadoop/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/docker/hoodie/hadoop/prestobase/pom.xml b/docker/hoodie/hadoop/prestobase/pom.xml
index cb63e87..5dc488c 100644
--- a/docker/hoodie/hadoop/prestobase/pom.xml
+++ b/docker/hoodie/hadoop/prestobase/pom.xml
@@ -20,7 +20,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/spark_base/pom.xml b/docker/hoodie/hadoop/spark_base/pom.xml
index bc1e776..c54ab85 100644
--- a/docker/hoodie/hadoop/spark_base/pom.xml
+++ b/docker/hoodie/hadoop/spark_base/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/sparkadhoc/pom.xml b/docker/hoodie/hadoop/sparkadhoc/pom.xml
index d2972b5..deb7c25 100644
--- a/docker/hoodie/hadoop/sparkadhoc/pom.xml
+++ b/docker/hoodie/hadoop/sparkadhoc/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/sparkmaster/pom.xml b/docker/hoodie/hadoop/sparkmaster/pom.xml
index a2d7919..1835178 100644
--- a/docker/hoodie/hadoop/sparkmaster/pom.xml
+++ b/docker/hoodie/hadoop/sparkmaster/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/docker/hoodie/hadoop/sparkworker/pom.xml b/docker/hoodie/hadoop/sparkworker/pom.xml
index df1fae7..ec69399 100644
--- a/docker/hoodie/hadoop/sparkworker/pom.xml
+++ b/docker/hoodie/hadoop/sparkworker/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi-hadoop-docker</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
diff --git a/hudi-aws/pom.xml b/hudi-aws/pom.xml
index 1e9c94c..c100d50 100644
--- a/hudi-aws/pom.xml
+++ b/hudi-aws/pom.xml
@@ -19,12 +19,12 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-aws</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-aws</name>
<packaging>jar</packaging>
diff --git a/hudi-cli/pom.xml b/hudi-cli/pom.xml
index 7a62f06..1da3402 100644
--- a/hudi-cli/pom.xml
+++ b/hudi-cli/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-client/hudi-client-common/pom.xml b/hudi-client/hudi-client-common/pom.xml
index 5c265ed..b453907 100644
--- a/hudi-client/hudi-client-common/pom.xml
+++ b/hudi-client/hudi-client-common/pom.xml
@@ -19,12 +19,12 @@
<parent>
<artifactId>hudi-client</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-client-common</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-client-common</name>
<packaging>jar</packaging>
diff --git a/hudi-client/hudi-flink-client/pom.xml b/hudi-client/hudi-flink-client/pom.xml
index 1605db7..245d233 100644
--- a/hudi-client/hudi-flink-client/pom.xml
+++ b/hudi-client/hudi-flink-client/pom.xml
@@ -19,12 +19,12 @@
<parent>
<artifactId>hudi-client</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-flink-client</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-flink-client</name>
<packaging>jar</packaging>
diff --git a/hudi-client/hudi-java-client/pom.xml b/hudi-client/hudi-java-client/pom.xml
index 9537847..1ed65b3 100644
--- a/hudi-client/hudi-java-client/pom.xml
+++ b/hudi-client/hudi-java-client/pom.xml
@@ -19,12 +19,12 @@
<parent>
<artifactId>hudi-client</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-java-client</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-java-client</name>
<packaging>jar</packaging>
diff --git a/hudi-client/hudi-spark-client/pom.xml b/hudi-client/hudi-spark-client/pom.xml
index 00ff2ca..5f64ad6 100644
--- a/hudi-client/hudi-spark-client/pom.xml
+++ b/hudi-client/hudi-spark-client/pom.xml
@@ -19,12 +19,12 @@
<parent>
<artifactId>hudi-client</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-spark-client</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-spark-client</name>
<packaging>jar</packaging>
diff --git a/hudi-client/pom.xml b/hudi-client/pom.xml
index 7f67cee..803d663 100644
--- a/hudi-client/pom.xml
+++ b/hudi-client/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-common/pom.xml b/hudi-common/pom.xml
index 5e50d50..a42bbee 100644
--- a/hudi-common/pom.xml
+++ b/hudi-common/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-examples/pom.xml b/hudi-examples/pom.xml
index fdc743b..3340830 100644
--- a/hudi-examples/pom.xml
+++ b/hudi-examples/pom.xml
@@ -20,7 +20,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-flink/pom.xml b/hudi-flink/pom.xml
index b5090eb..d7246f4 100644
--- a/hudi-flink/pom.xml
+++ b/hudi-flink/pom.xml
@@ -21,7 +21,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-hadoop-mr/pom.xml b/hudi-hadoop-mr/pom.xml
index 31b5e7b..34c40ec 100644
--- a/hudi-hadoop-mr/pom.xml
+++ b/hudi-hadoop-mr/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-integ-test/pom.xml b/hudi-integ-test/pom.xml
index 2469043..3ba7634 100644
--- a/hudi-integ-test/pom.xml
+++ b/hudi-integ-test/pom.xml
@@ -21,7 +21,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>hudi-integ-test</artifactId>
diff --git a/hudi-kafka-connect/pom.xml b/hudi-kafka-connect/pom.xml
index ca0bbe9..84911ca 100644
--- a/hudi-kafka-connect/pom.xml
+++ b/hudi-kafka-connect/pom.xml
@@ -19,13 +19,13 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-kafka-connect</artifactId>
<description>Kafka Connect Sink Connector for Hudi</description>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<packaging>jar</packaging>
<properties>
diff --git a/hudi-spark-datasource/hudi-spark-common/pom.xml b/hudi-spark-datasource/hudi-spark-common/pom.xml
index d64a222..f331503 100644
--- a/hudi-spark-datasource/hudi-spark-common/pom.xml
+++ b/hudi-spark-datasource/hudi-spark-common/pom.xml
@@ -17,12 +17,12 @@
<parent>
<artifactId>hudi-spark-datasource</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-spark-common_${scala.binary.version}</name>
<packaging>jar</packaging>
diff --git a/hudi-spark-datasource/hudi-spark/pom.xml b/hudi-spark-datasource/hudi-spark/pom.xml
index c46cc09..21e063f 100644
--- a/hudi-spark-datasource/hudi-spark/pom.xml
+++ b/hudi-spark-datasource/hudi-spark/pom.xml
@@ -19,12 +19,12 @@
<parent>
<artifactId>hudi-spark-datasource</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-spark_${scala.binary.version}</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-spark_${scala.binary.version}</name>
<packaging>jar</packaging>
diff --git a/hudi-spark-datasource/hudi-spark2/pom.xml b/hudi-spark-datasource/hudi-spark2/pom.xml
index 0dc8e21..89482ed 100644
--- a/hudi-spark-datasource/hudi-spark2/pom.xml
+++ b/hudi-spark-datasource/hudi-spark2/pom.xml
@@ -17,12 +17,12 @@
<parent>
<artifactId>hudi-spark-datasource</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-spark2_${scala.binary.version}</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-spark2_${scala.binary.version}</name>
<packaging>jar</packaging>
diff --git a/hudi-spark-datasource/hudi-spark3/pom.xml b/hudi-spark-datasource/hudi-spark3/pom.xml
index a2a94a3..f819139 100644
--- a/hudi-spark-datasource/hudi-spark3/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3/pom.xml
@@ -17,12 +17,12 @@
<parent>
<artifactId>hudi-spark-datasource</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-spark3_2.12</artifactId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<name>hudi-spark3_2.12</name>
<packaging>jar</packaging>
diff --git a/hudi-spark-datasource/pom.xml b/hudi-spark-datasource/pom.xml
index af94678..b84235a 100644
--- a/hudi-spark-datasource/pom.xml
+++ b/hudi-spark-datasource/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-sync/hudi-dla-sync/pom.xml b/hudi-sync/hudi-dla-sync/pom.xml
index f23c14a..66c9307 100644
--- a/hudi-sync/hudi-dla-sync/pom.xml
+++ b/hudi-sync/hudi-dla-sync/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
diff --git a/hudi-sync/hudi-hive-sync/pom.xml b/hudi-sync/hudi-hive-sync/pom.xml
index 7bf5567..bdda6d4 100644
--- a/hudi-sync/hudi-hive-sync/pom.xml
+++ b/hudi-sync/hudi-hive-sync/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
diff --git a/hudi-sync/hudi-sync-common/pom.xml b/hudi-sync/hudi-sync-common/pom.xml
index f1778d5..f7beb40 100644
--- a/hudi-sync/hudi-sync-common/pom.xml
+++ b/hudi-sync/hudi-sync-common/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-sync/pom.xml b/hudi-sync/pom.xml
index 4718a4e..a9b69e8 100644
--- a/hudi-sync/pom.xml
+++ b/hudi-sync/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-timeline-service/pom.xml b/hudi-timeline-service/pom.xml
index ab3fd1d..cda3bb1 100644
--- a/hudi-timeline-service/pom.xml
+++ b/hudi-timeline-service/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/hudi-utilities/pom.xml b/hudi-utilities/pom.xml
index 65183a1..ce86449 100644
--- a/hudi-utilities/pom.xml
+++ b/hudi-utilities/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-flink-bundle/pom.xml b/packaging/hudi-flink-bundle/pom.xml
index 8a06063..7fd6eb1 100644
--- a/packaging/hudi-flink-bundle/pom.xml
+++ b/packaging/hudi-flink-bundle/pom.xml
@@ -21,7 +21,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-hadoop-mr-bundle/pom.xml b/packaging/hudi-hadoop-mr-bundle/pom.xml
index 1f513b0..0df22a8 100644
--- a/packaging/hudi-hadoop-mr-bundle/pom.xml
+++ b/packaging/hudi-hadoop-mr-bundle/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-hive-sync-bundle/pom.xml b/packaging/hudi-hive-sync-bundle/pom.xml
index e2afa3d..c0ade3a 100644
--- a/packaging/hudi-hive-sync-bundle/pom.xml
+++ b/packaging/hudi-hive-sync-bundle/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-integ-test-bundle/pom.xml b/packaging/hudi-integ-test-bundle/pom.xml
index 500691c..b328569 100644
--- a/packaging/hudi-integ-test-bundle/pom.xml
+++ b/packaging/hudi-integ-test-bundle/pom.xml
@@ -17,7 +17,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-kafka-connect-bundle/pom.xml b/packaging/hudi-kafka-connect-bundle/pom.xml
index 1c77c0a..fead2fe 100644
--- a/packaging/hudi-kafka-connect-bundle/pom.xml
+++ b/packaging/hudi-kafka-connect-bundle/pom.xml
@@ -20,7 +20,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-presto-bundle/pom.xml b/packaging/hudi-presto-bundle/pom.xml
index 5a91625..266fd40 100644
--- a/packaging/hudi-presto-bundle/pom.xml
+++ b/packaging/hudi-presto-bundle/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-spark-bundle/pom.xml b/packaging/hudi-spark-bundle/pom.xml
index 2307c75..2d1dc62 100644
--- a/packaging/hudi-spark-bundle/pom.xml
+++ b/packaging/hudi-spark-bundle/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-timeline-server-bundle/pom.xml b/packaging/hudi-timeline-server-bundle/pom.xml
index e570619..13e57a8 100644
--- a/packaging/hudi-timeline-server-bundle/pom.xml
+++ b/packaging/hudi-timeline-server-bundle/pom.xml
@@ -21,7 +21,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/packaging/hudi-utilities-bundle/pom.xml b/packaging/hudi-utilities-bundle/pom.xml
index 7f1efd7..bd761be 100644
--- a/packaging/hudi-utilities-bundle/pom.xml
+++ b/packaging/hudi-utilities-bundle/pom.xml
@@ -19,7 +19,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pom.xml b/pom.xml
index 09dbe47..00e9580 100644
--- a/pom.xml
+++ b/pom.xml
@@ -29,7 +29,7 @@
<groupId>org.apache.hudi</groupId>
<artifactId>hudi</artifactId>
<packaging>pom</packaging>
- <version>0.10.0-rc2</version>
+ <version>0.10.0-rc3</version>
<description>Apache Hudi brings stream style processing on big data</description>
<url>https://github.com/apache/hudi</url>
<name>Hudi</name>
[hudi] 07/26: [HUDI-2881] Compact the file group with larger log files to reduce write amplification (#4152)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 7d2c2489025fe5b183367b4f4dfbf64804d8c33b
Author: Shawy Geng <ge...@gmail.com>
AuthorDate: Thu Dec 2 09:41:04 2021 +0800
[HUDI-2881] Compact the file group with larger log files to reduce write amplification (#4152)
(cherry picked from commit 5284730175df4637eee43b179c774606b07a10a9)
---
.../java/org/apache/hudi/config/HoodieCompactionConfig.java | 11 +++++++++++
.../main/java/org/apache/hudi/config/HoodieWriteConfig.java | 4 ++++
.../compact/strategy/LogFileSizeBasedCompactionStrategy.java | 9 +++++++--
.../action/compact/strategy/TestHoodieCompactionStrategy.java | 9 +++++----
4 files changed, 27 insertions(+), 6 deletions(-)
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 e11d060..fbe31b0 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
@@ -171,6 +171,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<Long> 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<String> COMPACTION_STRATEGY = ConfigProperty
.key("hoodie.compaction.strategy")
.defaultValue(LogFileSizeBasedCompactionStrategy.class.getName())
@@ -598,6 +604,11 @@ public class HoodieCompactionConfig extends HoodieConfig {
return this;
}
+ public Builder withLogFileSizeThresholdBasedCompaction(long logFileSizeThreshold) {
+ compactionConfig.setValue(COMPACTION_LOG_FILE_SIZE_THRESHOLD, String.valueOf(logFileSizeThreshold));
+ return this;
+ }
+
public Builder withCommitsArchivalBatchSize(int batchSize) {
compactionConfig.setValue(COMMITS_ARCHIVAL_BATCH_SIZE, String.valueOf(batchSize));
return this;
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index 9ae22e8..b49108f 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
@@ -1133,6 +1133,10 @@ public class HoodieWriteConfig extends HoodieConfig {
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);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/LogFileSizeBasedCompactionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/LogFileSizeBasedCompactionStrategy.java
index fe4b474..c165141 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/LogFileSizeBasedCompactionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/strategy/LogFileSizeBasedCompactionStrategy.java
@@ -27,7 +27,8 @@ import java.util.List;
import java.util.stream.Collectors;
/**
- * LogFileSizeBasedCompactionStrategy orders the compactions based on the total log files size and limits the
+ * LogFileSizeBasedCompactionStrategy orders the compactions based on the total log files size,
+ * filters the file group which log files size is greater than the threshold and limits the
* compactions within a configured IO bound.
*
* @see BoundedIOCompactionStrategy
@@ -39,8 +40,12 @@ public class LogFileSizeBasedCompactionStrategy extends BoundedIOCompactionStrat
@Override
public List<HoodieCompactionOperation> orderAndFilter(HoodieWriteConfig writeConfig,
List<HoodieCompactionOperation> operations, List<HoodieCompactionPlan> pendingCompactionPlans) {
+ // Filter the file group which log files size is greater than the threshold in bytes.
// Order the operations based on the reverse size of the logs and limit them by the IO
- return super.orderAndFilter(writeConfig, operations.stream().sorted(this).collect(Collectors.toList()),
+ long threshold = writeConfig.getCompactionLogFileSizeThreshold();
+ return super.orderAndFilter(writeConfig, operations.stream()
+ .filter(e -> e.getMetrics().getOrDefault(TOTAL_LOG_FILE_SIZE, 0d) >= threshold)
+ .sorted(this).collect(Collectors.toList()),
pendingCompactionPlans);
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java
index 823eac4..dee1fad 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java
@@ -99,19 +99,20 @@ public class TestHoodieCompactionStrategy {
sizesMap.put(90 * MB, Collections.singletonList(1024 * MB));
LogFileSizeBasedCompactionStrategy strategy = new LogFileSizeBasedCompactionStrategy();
HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder().withPath("/tmp").withCompactionConfig(
- HoodieCompactionConfig.newBuilder().withCompactionStrategy(strategy).withTargetIOPerCompactionInMB(400).build())
+ HoodieCompactionConfig.newBuilder().withCompactionStrategy(strategy).withTargetIOPerCompactionInMB(1205)
+ .withLogFileSizeThresholdBasedCompaction(100 * 1024 * 1024).build())
.build();
List<HoodieCompactionOperation> operations = createCompactionOperations(writeConfig, sizesMap);
List<HoodieCompactionOperation> returned = strategy.orderAndFilter(writeConfig, operations, new ArrayList<>());
assertTrue(returned.size() < operations.size(),
"LogFileSizeBasedCompactionStrategy should have resulted in fewer compactions");
- assertEquals(1, returned.size(), "LogFileSizeBasedCompactionStrategy should have resulted in 1 compaction");
+ assertEquals(2, returned.size(), "LogFileSizeBasedCompactionStrategy should have resulted in 2 compaction");
// Total size of all the log files
Long returnedSize = returned.stream().map(s -> s.getMetrics().get(BoundedIOCompactionStrategy.TOTAL_IO_MB))
.map(Double::longValue).reduce(Long::sum).orElse(0L);
- assertEquals(1204, (long) returnedSize,
- "Should chose the first 2 compactions which should result in a total IO of 690 MB");
+ assertEquals(1594, (long) returnedSize,
+ "Should chose the first 2 compactions which should result in a total IO of 1594 MB");
}
@Test
[hudi] 12/26: [minor] Refactor write profile to always generate fs view (#4198)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit fe2aae4cea9617cbb469f7b4c6aa072d4c2f3f55
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Fri Dec 3 11:38:29 2021 +0800
[minor] Refactor write profile to always generate fs view (#4198)
(cherry picked from commit f74b3d12aa89a96daafe43c69d37d816b9f4018b)
---
.../partitioner/profile/DeltaWriteProfile.java | 4 +--
.../sink/partitioner/profile/WriteProfile.java | 29 +++++++++++++++-------
2 files changed, 22 insertions(+), 11 deletions(-)
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
index 1f08c5a..e1c890c 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java
@@ -50,7 +50,7 @@ public class DeltaWriteProfile extends WriteProfile {
List<SmallFile> smallFileLocations = new ArrayList<>();
// Init here since this class (and member variables) might not have been initialized
- HoodieTimeline commitTimeline = table.getCompletedCommitsTimeline();
+ HoodieTimeline commitTimeline = metaClient.getCommitsTimeline().filterCompletedInstants();
// Find out all eligible small file slices
if (!commitTimeline.empty()) {
@@ -90,7 +90,7 @@ public class DeltaWriteProfile extends WriteProfile {
}
protected SyncableFileSystemView getFileSystemView() {
- return (SyncableFileSystemView) this.table.getSliceView();
+ return (SyncableFileSystemView) getTable().getSliceView();
}
private long getTotalFileSize(FileSlice fileSlice) {
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
index 98eb29c..d13162f 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
@@ -32,6 +32,7 @@ import org.apache.hudi.sink.partitioner.BucketAssigner;
import org.apache.hudi.table.HoodieFlinkTable;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.commit.SmallFile;
+import org.apache.hudi.util.StreamerUtil;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.fs.Path;
@@ -67,9 +68,9 @@ public class WriteProfile {
private final Path basePath;
/**
- * The hoodie table.
+ * The meta client.
*/
- protected final HoodieTable<?, ?, ?, ?> table;
+ protected final HoodieTableMetaClient metaClient;
/**
* The average record size.
@@ -97,12 +98,18 @@ public class WriteProfile {
*/
private final Map<String, HoodieCommitMetadata> metadataCache;
+ /**
+ * The engine context.
+ */
+ private final HoodieFlinkEngineContext context;
+
public WriteProfile(HoodieWriteConfig config, HoodieFlinkEngineContext context) {
this.config = config;
+ this.context = context;
this.basePath = new Path(config.getBasePath());
this.smallFilesMap = new HashMap<>();
this.recordsPerBucket = config.getCopyOnWriteInsertSplitSize();
- this.table = HoodieFlinkTable.create(config, context);
+ this.metaClient = StreamerUtil.createMetaClient(config.getBasePath(), context.getHadoopConf().get());
this.metadataCache = new HashMap<>();
// profile the record statistics on construction
recordProfile();
@@ -117,7 +124,11 @@ public class WriteProfile {
}
public HoodieTableMetaClient getMetaClient() {
- return this.table.getMetaClient();
+ return this.metaClient;
+ }
+
+ protected HoodieTable<?, ?, ?, ?> getTable() {
+ return HoodieFlinkTable.create(config, context);
}
/**
@@ -127,7 +138,7 @@ public class WriteProfile {
private long averageBytesPerRecord() {
long avgSize = config.getCopyOnWriteRecordSizeEstimate();
long fileSizeThreshold = (long) (config.getRecordSizeEstimationThreshold() * config.getParquetSmallFileLimit());
- HoodieTimeline commitTimeline = table.getMetaClient().getCommitsTimeline().filterCompletedInstants();
+ HoodieTimeline commitTimeline = metaClient.getCommitsTimeline().filterCompletedInstants();
if (!commitTimeline.empty()) {
// Go over the reverse ordered commits to get a more recent estimate of average record size.
Iterator<HoodieInstant> instants = commitTimeline.getReverseOrderedInstants().iterator();
@@ -175,7 +186,7 @@ public class WriteProfile {
// smallFiles only for partitionPath
List<SmallFile> smallFileLocations = new ArrayList<>();
- HoodieTimeline commitTimeline = table.getMetaClient().getCommitsTimeline().filterCompletedInstants();
+ HoodieTimeline commitTimeline = metaClient.getCommitsTimeline().filterCompletedInstants();
if (!commitTimeline.empty()) { // if we have some commits
HoodieInstant latestCommitTime = commitTimeline.lastInstant().get();
@@ -198,7 +209,7 @@ public class WriteProfile {
}
protected SyncableFileSystemView getFileSystemView() {
- return (SyncableFileSystemView) HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) table.getContext()).getBaseFileOnlyView();
+ return (SyncableFileSystemView) getTable().getBaseFileOnlyView();
}
/**
@@ -231,9 +242,9 @@ public class WriteProfile {
// already reloaded
return;
}
- this.table.getMetaClient().reloadActiveTimeline();
+ this.metaClient.reloadActiveTimeline();
recordProfile();
- cleanMetadataCache(this.table.getMetaClient().getCommitsTimeline().filterCompletedInstants().getInstants());
+ cleanMetadataCache(this.metaClient.getCommitsTimeline().filterCompletedInstants().getInstants());
this.smallFilesMap.clear();
this.reloadedCheckpointId = checkpointId;
}
[hudi] 14/26: [MINOR] use catalog schema if can not find table schema (#4182)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 1bf13d1a0f15e64433f915b32066caa8b7d76174
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Fri Dec 3 16:37:13 2021 +0800
[MINOR] use catalog schema if can not find table schema (#4182)
(cherry picked from commit ca427240c026bc6ba6e19b9b8b0a069a6330147a)
---
.../org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala
index 7808122..fdf1b06 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala
@@ -117,7 +117,7 @@ class HoodieCatalogTable(val spark: SparkSession, val table: CatalogTable) exten
* Make StructField nullable.
*/
lazy val tableSchema: StructType = {
- val originSchema = getTableSqlSchema(metaClient, includeMetadataFields = true).get
+ val originSchema = getTableSqlSchema(metaClient, includeMetadataFields = true).getOrElse(table.schema)
StructType(originSchema.map(_.copy(nullable = true)))
}
[hudi] 18/26: [HUDI-2894][HUDI-2905] Metadata table - avoiding key lookup failures on base files over S3 (#4185)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a49e5ac7160a5937ebfe2f72deb36dc0ab119f2e
Author: Manoj Govindassamy <ma...@gmail.com>
AuthorDate: Fri Dec 3 11:18:10 2021 -0800
[HUDI-2894][HUDI-2905] Metadata table - avoiding key lookup failures on base files over S3 (#4185)
- Fetching partition files or all partitions from the metadata table is failing
when run over S3. Metadata table uses HFile format for the base files and the
record lookup uses HFile.Reader and HFileScanner interfaces to get records by
partition keys. When the backing storage is S3, this record lookup from HFiles
is failing with IOException, in turn failing the caller commit/update operations.
- Metadata table looks up HFile records with positional read enabled so as to
perform better for random lookups. But this positional read key lookup is
returning with partial read sizes over S3 leading to HFile scanner throwing
IOException. This doesn't happen over HDFS. Metadata table though uses the HFile
for random key lookups, the positional read is not mandatory as we sort the keys
when doing a lookup for multiple keys.
- The fix is to disable HFile positional read for all HFile scanner based
key lookups.
(cherry picked from commit 383d5edc169b79c4022f81a1580ab1bc2afebc30)
---
.../apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java | 8 ++++----
.../main/java/org/apache/hudi/io/storage/HoodieHFileReader.java | 2 +-
2 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java b/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java
index d4a77b0..3700d01 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java
@@ -291,13 +291,13 @@ public class HFileBootstrapIndex extends BootstrapIndex {
@Override
public List<String> getIndexedPartitionPaths() {
- HFileScanner scanner = partitionIndexReader().getScanner(true, true);
+ HFileScanner scanner = partitionIndexReader().getScanner(true, false);
return getAllKeys(scanner, HFileBootstrapIndex::getPartitionFromKey);
}
@Override
public List<HoodieFileGroupId> getIndexedFileGroupIds() {
- HFileScanner scanner = fileIdIndexReader().getScanner(true, true);
+ HFileScanner scanner = fileIdIndexReader().getScanner(true, false);
return getAllKeys(scanner, HFileBootstrapIndex::getFileGroupFromKey);
}
@@ -319,7 +319,7 @@ public class HFileBootstrapIndex extends BootstrapIndex {
@Override
public List<BootstrapFileMapping> getSourceFileMappingForPartition(String partition) {
try {
- HFileScanner scanner = partitionIndexReader().getScanner(true, true);
+ HFileScanner scanner = partitionIndexReader().getScanner(true, false);
KeyValue keyValue = new KeyValue(Bytes.toBytes(getPartitionKey(partition)), new byte[0], new byte[0],
HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put, new byte[0]);
if (scanner.seekTo(keyValue) == 0) {
@@ -352,7 +352,7 @@ public class HFileBootstrapIndex extends BootstrapIndex {
List<HoodieFileGroupId> fileGroupIds = new ArrayList<>(ids);
Collections.sort(fileGroupIds);
try {
- HFileScanner scanner = fileIdIndexReader().getScanner(true, true);
+ HFileScanner scanner = fileIdIndexReader().getScanner(true, false);
for (HoodieFileGroupId fileGroupId : fileGroupIds) {
KeyValue keyValue = new KeyValue(Bytes.toBytes(getFileGroupKey(fileGroupId)), new byte[0], new byte[0],
HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put, new byte[0]);
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileReader.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileReader.java
index 7b80d1a..e3e38ec 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileReader.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileReader.java
@@ -246,7 +246,7 @@ public class HoodieHFileReader<R extends IndexedRecord> implements HoodieFileRea
synchronized (this) {
if (keyScanner == null) {
- keyScanner = reader.getScanner(false, true);
+ keyScanner = reader.getScanner(false, false);
}
if (keyScanner.seekTo(kv) == 0) {
[hudi] 25/26: [HUDI-2935] Remove special casing of clustering in deltastreamer checkpoint retrival (#4216)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 40879bca148932c9ad5a13e761cb34f99bc68932
Author: vinoth chandar <vi...@users.noreply.github.com>
AuthorDate: Sat Dec 4 01:16:11 2021 -0800
[HUDI-2935] Remove special casing of clustering in deltastreamer checkpoint retrival (#4216)
- We now seek backwards to find the checkpoint
- No need to return empty anymore
(cherry picked from commit 36b69d80332468439d9e1674b7d1845198379152)
---
.../main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java | 3 ---
1 file changed, 3 deletions(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
index 1baa68f..07f37a7 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
@@ -336,9 +336,6 @@ public class DeltaSync implements Serializable {
} else if (!StringUtils.isNullOrEmpty(commitMetadata.getMetadata(CHECKPOINT_KEY))) {
//if previous checkpoint is an empty string, skip resume use Option.empty()
resumeCheckpointStr = Option.of(commitMetadata.getMetadata(CHECKPOINT_KEY));
- } else if (commitMetadata.getOperationType() == WriteOperationType.CLUSTER) {
- // incase of CLUSTER commit, no checkpoint will be available in metadata.
- resumeCheckpointStr = Option.empty();
} else if (HoodieTimeline.compareTimestamps(HoodieTimeline.FULL_BOOTSTRAP_INSTANT_TS,
HoodieTimeline.LESSER_THAN, lastCommit.get().getTimestamp())) {
// if previous commit metadata did not have the checkpoint key, try traversing previous commits until we find one.
[hudi] 03/26: [HUDI-2891] Fix write configs for Java engine in Kafka Connect Sink (#4161)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit c38ebbc2353fccf9ca91b21307d316bae27e9074
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Tue Nov 30 06:45:50 2021 -0800
[HUDI-2891] Fix write configs for Java engine in Kafka Connect Sink (#4161)
(cherry picked from commit ea009b55a385564f76beb99468f363edd3c4828a)
---
hudi-kafka-connect/demo/config-sink-hive.json | 1 -
hudi-kafka-connect/demo/config-sink.json | 1 -
.../org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java | 2 ++
.../test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java | 2 ++
4 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/hudi-kafka-connect/demo/config-sink-hive.json b/hudi-kafka-connect/demo/config-sink-hive.json
index bf7e998..214fd18 100644
--- a/hudi-kafka-connect/demo/config-sink-hive.json
+++ b/hudi-kafka-connect/demo/config-sink-hive.json
@@ -10,7 +10,6 @@
"topics": "hudi-test-topic",
"hoodie.table.name": "hudi-test-topic",
"hoodie.table.type": "MERGE_ON_READ",
- "hoodie.metadata.enable": "false",
"hoodie.base.path": "hdfs://namenode:8020/user/hive/warehouse/hudi-test-topic",
"hoodie.datasource.write.recordkey.field": "volume",
"hoodie.datasource.write.partitionpath.field": "date",
diff --git a/hudi-kafka-connect/demo/config-sink.json b/hudi-kafka-connect/demo/config-sink.json
index 9d1aedf..105c4d4 100644
--- a/hudi-kafka-connect/demo/config-sink.json
+++ b/hudi-kafka-connect/demo/config-sink.json
@@ -10,7 +10,6 @@
"topics": "hudi-test-topic",
"hoodie.table.name": "hudi-test-topic",
"hoodie.table.type": "MERGE_ON_READ",
- "hoodie.metadata.enable": "false",
"hoodie.base.path": "file:///tmp/hoodie/hudi-test-topic",
"hoodie.datasource.write.recordkey.field": "volume",
"hoodie.datasource.write.partitionpath.field": "date",
diff --git a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
index 18c1835..5bac1f0 100644
--- a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
+++ b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
@@ -22,6 +22,7 @@ import org.apache.hudi.client.HoodieJavaWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.common.HoodieJavaEngineContext;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieAvroPayload;
import org.apache.hudi.common.util.ReflectionUtils;
@@ -74,6 +75,7 @@ public class KafkaConnectWriterProvider implements ConnectWriterProvider<WriteSt
// Create the write client to write some records in
writeConfig = HoodieWriteConfig.newBuilder()
+ .withEngineType(EngineType.JAVA)
.withProperties(connectConfigs.getProps())
.withFileIdPrefixProviderClassName(KafkaConnectFileIdPrefixProvider.class.getName())
.withProps(Collections.singletonMap(
diff --git a/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java b/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java
index b0dcf38..458c79a 100644
--- a/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java
+++ b/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java
@@ -20,6 +20,7 @@ package org.apache.hudi.writers;
import org.apache.hudi.client.HoodieJavaWriteClient;
import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.Option;
@@ -62,6 +63,7 @@ public class TestBufferedConnectWriter {
configs = KafkaConnectConfigs.newBuilder().build();
schemaProvider = new TestAbstractConnectWriter.TestSchemaProvider();
writeConfig = HoodieWriteConfig.newBuilder()
+ .withEngineType(EngineType.JAVA)
.withPath("/tmp")
.withSchema(schemaProvider.getSourceSchema().toString())
.build();
[hudi] 15/26: [HUDI-2902] Fixing populate meta fields with Hfile writers and Disabling virtual keys by default for metadata table (#4194)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 7de5d1760191128c86ef40ac2f9cb57e87328e3e
Author: Sivabalan Narayanan <si...@uber.com>
AuthorDate: Fri Dec 3 07:20:21 2021 -0500
[HUDI-2902] Fixing populate meta fields with Hfile writers and Disabling virtual keys by default for metadata table (#4194)
(cherry picked from commit e483f7c776ed53015d08b78fc33729892dc2143c)
---
.../apache/hudi/io/HoodieSortedMergeHandle.java | 3 +-
.../hudi/io/storage/HoodieFileWriterFactory.java | 2 +-
.../apache/hudi/io/storage/HoodieHFileWriter.java | 14 +-
.../io/storage/TestHoodieHFileReaderWriter.java | 55 ++++++--
.../resources/exampleSchemaWithMetaFields.avsc | 66 +++++++++
.../functional/TestHoodieBackedMetadata.java | 157 +++++++++++++++++++++
.../client/functional/TestHoodieMetadataBase.java | 2 +-
.../hudi/common/config/HoodieMetadataConfig.java | 12 +-
.../HoodieMetadataMergedLogRecordReader.java | 4 +-
...inuousModeWithMultipleWriters.COPY_ON_WRITE.zip | Bin 2592485 -> 2592484 bytes
...inuousModeWithMultipleWriters.MERGE_ON_READ.zip | Bin 3015940 -> 3015939 bytes
11 files changed, 287 insertions(+), 28 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
index 606e63a..533611d 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
@@ -27,6 +27,7 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.keygen.KeyGenUtils;
import org.apache.hudi.table.HoodieTable;
import org.apache.avro.generic.GenericRecord;
@@ -72,7 +73,7 @@ public class HoodieSortedMergeHandle<T extends HoodieRecordPayload, I, K, O> ext
*/
@Override
public void write(GenericRecord oldRecord) {
- String key = oldRecord.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
+ String key = KeyGenUtils.getRecordKeyFromGenericRecord(oldRecord, keyGeneratorOpt);
// To maintain overall sorted order across updates and inserts, write any new inserts whose keys are less than
// the oldRecord's key.
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java
index c6d4540..0b6afd4 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java
@@ -89,7 +89,7 @@ public class HoodieFileWriterFactory {
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 <T extends HoodieRecordPayload, R extends IndexedRecord> HoodieFileWriter<R> newOrcFileWriter(
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 d7e01c5..a719bcb 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
@@ -62,6 +62,7 @@ public class HoodieHFileWriter<T extends HoodieRecordPayload, R extends IndexedR
private final long maxFileSize;
private final String instantTime;
private final TaskContextSupplier taskContextSupplier;
+ private final boolean populateMetaFields;
private HFile.Writer writer;
private String minRecordKey;
private String maxRecordKey;
@@ -70,7 +71,7 @@ public class HoodieHFileWriter<T extends HoodieRecordPayload, R extends IndexedR
private static String DROP_BEHIND_CACHE_COMPACTION_KEY = "hbase.hfile.drop.behind.compaction";
public HoodieHFileWriter(String instantTime, Path file, HoodieHFileConfig hfileConfig, Schema schema,
- TaskContextSupplier taskContextSupplier) throws IOException {
+ TaskContextSupplier taskContextSupplier, boolean populateMetaFields) throws IOException {
Configuration conf = FSUtils.registerFileSystem(file, hfileConfig.getHadoopConf());
this.file = HoodieWrapperFileSystem.convertToHoodiePath(file, conf);
@@ -84,6 +85,7 @@ public class HoodieHFileWriter<T extends HoodieRecordPayload, R extends IndexedR
this.maxFileSize = hfileConfig.getMaxFileSize();
this.instantTime = instantTime;
this.taskContextSupplier = taskContextSupplier;
+ this.populateMetaFields = populateMetaFields;
HFileContext context = new HFileContextBuilder().withBlockSize(hfileConfig.getBlockSize())
.withCompression(hfileConfig.getCompressionAlgorithm())
@@ -104,9 +106,13 @@ public class HoodieHFileWriter<T extends HoodieRecordPayload, R extends IndexedR
@Override
public void writeAvroWithMetadata(R avroRecord, HoodieRecord record) throws IOException {
- prepRecordWithMetadata(avroRecord, record, instantTime,
- taskContextSupplier.getPartitionIdSupplier().get(), recordIndex, file.getName());
- writeAvro(record.getRecordKey(), (IndexedRecord) avroRecord);
+ if (populateMetaFields) {
+ prepRecordWithMetadata(avroRecord, record, instantTime,
+ taskContextSupplier.getPartitionIdSupplier().get(), recordIndex, file.getName());
+ writeAvro(record.getRecordKey(), (IndexedRecord) avroRecord);
+ } else {
+ writeAvro(record.getRecordKey(), (IndexedRecord) avroRecord);
+ }
}
@Override
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java
index c087607..86a0886 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java
@@ -22,6 +22,9 @@ import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.bloom.BloomFilterFactory;
import org.apache.hudi.common.bloom.BloomFilterTypeCode;
import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
@@ -34,19 +37,24 @@ import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.util.Pair;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
import static org.apache.hudi.common.testutils.FileSystemTestUtils.RANDOM;
import static org.apache.hudi.common.testutils.SchemaTestUtil.getSchemaFromResource;
@@ -55,6 +63,9 @@ import static org.apache.hudi.io.storage.HoodieHFileConfig.DROP_BEHIND_CACHE_COM
import static org.apache.hudi.io.storage.HoodieHFileConfig.HFILE_COMPARATOR;
import static org.apache.hudi.io.storage.HoodieHFileConfig.PREFETCH_ON_OPEN;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.when;
public class TestHoodieHFileReaderWriter {
@TempDir File tempDir;
@@ -73,21 +84,34 @@ public class TestHoodieHFileReaderWriter {
}
}
- private HoodieHFileWriter createHFileWriter(Schema avroSchema) throws Exception {
+ private static Stream<Arguments> populateMetaFieldsAndTestAvroWithMeta() {
+ return Arrays.stream(new Boolean[][] {
+ {true, true},
+ {false, true},
+ {true, false},
+ {false, false}
+ }).map(Arguments::of);
+ }
+
+ private HoodieHFileWriter createHFileWriter(Schema avroSchema, boolean populateMetaFields) throws Exception {
BloomFilter filter = BloomFilterFactory.createBloomFilter(1000, 0.00001, -1, BloomFilterTypeCode.SIMPLE.name());
Configuration conf = new Configuration();
TaskContextSupplier mockTaskContextSupplier = Mockito.mock(TaskContextSupplier.class);
+ Supplier<Integer> partitionSupplier = Mockito.mock(Supplier.class);
+ when(mockTaskContextSupplier.getPartitionIdSupplier()).thenReturn(partitionSupplier);
+ when(partitionSupplier.get()).thenReturn(10);
String instantTime = "000";
HoodieHFileConfig hoodieHFileConfig = new HoodieHFileConfig(conf, Compression.Algorithm.GZ, 1024 * 1024, 120 * 1024 * 1024,
PREFETCH_ON_OPEN, CACHE_DATA_IN_L1, DROP_BEHIND_CACHE_COMPACTION, filter, HFILE_COMPARATOR);
- return new HoodieHFileWriter(instantTime, filePath, hoodieHFileConfig, avroSchema, mockTaskContextSupplier);
+ return new HoodieHFileWriter(instantTime, filePath, hoodieHFileConfig, avroSchema, mockTaskContextSupplier, populateMetaFields);
}
- @Test
- public void testWriteReadHFile() throws Exception {
- Schema avroSchema = getSchemaFromResource(TestHoodieOrcReaderWriter.class, "/exampleSchema.avsc");
- HoodieHFileWriter writer = createHFileWriter(avroSchema);
+ @ParameterizedTest
+ @MethodSource("populateMetaFieldsAndTestAvroWithMeta")
+ public void testWriteReadHFile(boolean populateMetaFields, boolean testAvroWithMeta) throws Exception {
+ Schema avroSchema = getSchemaFromResource(TestHoodieOrcReaderWriter.class, "/exampleSchemaWithMetaFields.avsc");
+ HoodieHFileWriter writer = createHFileWriter(avroSchema, populateMetaFields);
List<String> keys = new ArrayList<>();
Map<String, GenericRecord> recordMap = new HashMap<>();
for (int i = 0; i < 100; i++) {
@@ -97,7 +121,13 @@ public class TestHoodieHFileReaderWriter {
keys.add(key);
record.put("time", Integer.toString(RANDOM.nextInt()));
record.put("number", i);
- writer.writeAvro(key, record);
+ if (testAvroWithMeta) {
+ writer.writeAvroWithMetadata(record, new HoodieRecord(new HoodieKey((String) record.get("_row_key"),
+ Integer.toString((Integer) record.get("number"))), new EmptyHoodieRecordPayload())); // payload does not matter. GenericRecord passed in is what matters
+ // only HoodieKey will be looked up from the 2nd arg(HoodieRecord).
+ } else {
+ writer.writeAvro(key, record);
+ }
recordMap.put(key, record);
}
writer.close();
@@ -109,8 +139,8 @@ public class TestHoodieHFileReaderWriter {
records.forEach(entry -> assertEquals(entry.getSecond(), recordMap.get(entry.getFirst())));
hoodieHFileReader.close();
- for (int i = 0; i < 20; i++) {
- int randomRowstoFetch = 5 + RANDOM.nextInt(50);
+ for (int i = 0; i < 2; i++) {
+ int randomRowstoFetch = 5 + RANDOM.nextInt(10);
Set<String> rowsToFetch = getRandomKeys(randomRowstoFetch, keys);
List<String> rowsList = new ArrayList<>(rowsToFetch);
Collections.sort(rowsList);
@@ -119,6 +149,11 @@ public class TestHoodieHFileReaderWriter {
assertEquals(result.size(), randomRowstoFetch);
result.forEach(entry -> {
assertEquals(entry.getSecond(), recordMap.get(entry.getFirst()));
+ if (populateMetaFields && testAvroWithMeta) {
+ assertNotNull(entry.getSecond().get(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ } else {
+ assertNull(entry.getSecond().get(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ }
});
hoodieHFileReader.close();
}
diff --git a/hudi-client/hudi-client-common/src/test/resources/exampleSchemaWithMetaFields.avsc b/hudi-client/hudi-client-common/src/test/resources/exampleSchemaWithMetaFields.avsc
new file mode 100644
index 0000000..c3fa822
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/test/resources/exampleSchemaWithMetaFields.avsc
@@ -0,0 +1,66 @@
+/*
+ * 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.
+ */
+{
+ "namespace": "example.schema",
+ "type": "record",
+ "name": "trip",
+ "fields": [
+ {
+ "name": "_hoodie_commit_time",
+ "type": ["null","string"],
+ "default":null
+ },
+ {
+ "name": "_hoodie_commit_seqno",
+ "type": ["null","string"],
+ "default":null
+ },
+ {
+ "name": "_hoodie_record_key",
+ "type": ["null","string"],
+ "default":null
+ },
+ {
+ "name": "_hoodie_partition_path",
+ "type": ["null","string"],
+ "default":null
+ },
+ {
+ "name": "_hoodie_file_name",
+ "type": ["null","string"],
+ "default":null
+ },
+ {
+ "name": "_hoodie_operation",
+ "type": ["null","string"],
+ "default":null
+ },
+ {
+ "name": "_row_key",
+ "type": "string"
+ },
+ {
+ "name": "time",
+ "type": "string"
+ },
+ {
+ "name": "number",
+ "type": ["int", "null"]
+ }
+ ]
+}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
index 73b7811..ed245da 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
@@ -18,6 +18,7 @@
package org.apache.hudi.client.functional;
+import org.apache.hudi.avro.model.HoodieMetadataRecord;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
@@ -29,6 +30,8 @@ import org.apache.hudi.common.fs.ConsistencyGuardConfig;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.metrics.Registry;
import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieCleaningPolicy;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieFileFormat;
@@ -41,6 +44,7 @@ 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.HoodieTableVersion;
+import org.apache.hudi.common.table.marker.MarkerType;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
@@ -60,12 +64,18 @@ import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieLockConfig;
import org.apache.hudi.config.HoodieStorageConfig;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.config.metrics.HoodieMetricsConfig;
+import org.apache.hudi.config.metrics.HoodieMetricsGraphiteConfig;
+import org.apache.hudi.config.metrics.HoodieMetricsJmxConfig;
import org.apache.hudi.exception.HoodieMetadataException;
import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.io.storage.HoodieHFileReader;
import org.apache.hudi.metadata.FileSystemBackedTableMetadata;
import org.apache.hudi.metadata.HoodieBackedTableMetadataWriter;
import org.apache.hudi.metadata.HoodieMetadataMetrics;
+import org.apache.hudi.metadata.HoodieMetadataPayload;
import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataKeyGenerator;
import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.metadata.SparkHoodieBackedTableMetadataWriter;
import org.apache.hudi.table.HoodieSparkTable;
@@ -75,9 +85,13 @@ import org.apache.hudi.table.upgrade.SparkUpgradeDowngradeHelper;
import org.apache.hudi.table.upgrade.UpgradeDowngrade;
import org.apache.hudi.testutils.MetadataMergeWriteStatus;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.util.Time;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
@@ -120,10 +134,12 @@ import static org.apache.hudi.common.model.WriteOperationType.DELETE;
import static org.apache.hudi.common.model.WriteOperationType.INSERT;
import static org.apache.hudi.common.model.WriteOperationType.UPSERT;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static org.apache.hudi.metadata.HoodieTableMetadata.METADATA_TABLE_NAME_SUFFIX;
import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -317,6 +333,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
/**
* Tests that table services in data table won't trigger table services in metadata table.
+ *
* @throws Exception
*/
@Test
@@ -346,6 +363,56 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
assertEquals(tableMetadata.getLatestCompactionTime().get(), "0000004001");
}
+
+ /**
+ * Tests that virtual key configs are honored in base files after compaction in metadata table.
+ *
+ * @throws Exception
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testVirtualKeysInBaseFiles(boolean populateMetaFields) throws Exception {
+ HoodieTableType tableType = MERGE_ON_READ;
+ init(tableType, false);
+ writeConfig = getWriteConfigBuilder(true, true, false)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .enableFullScan(true)
+ .enableMetrics(false)
+ .withPopulateMetaFields(populateMetaFields)
+ .withMaxNumDeltaCommitsBeforeCompaction(2)
+ .build()).build();
+ initWriteConfigAndMetatableWriter(writeConfig, true);
+
+ doWriteOperation(testTable, "0000001", INSERT);
+ doClean(testTable, "0000003", Arrays.asList("0000001"));
+ // this should have triggered compaction in metadata table
+ doWriteOperation(testTable, "0000004", UPSERT);
+
+ HoodieTableMetadata tableMetadata = metadata(writeConfig, context);
+ assertTrue(tableMetadata.getLatestCompactionTime().isPresent());
+ assertEquals(tableMetadata.getLatestCompactionTime().get(), "0000004001");
+
+ HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(metadataTableBasePath).build();
+ HoodieWriteConfig metadataTableWriteConfig = getMetadataWriteConfig(writeConfig);
+ metadataMetaClient.reloadActiveTimeline();
+
+ HoodieTable table = HoodieSparkTable.create(metadataTableWriteConfig, context, metadataMetaClient);
+ table.getHoodieView().sync();
+ List<FileSlice> fileSlices = table.getSliceView().getLatestFileSlices("files").collect(Collectors.toList());
+ HoodieBaseFile baseFile = fileSlices.get(0).getBaseFile().get();
+ HoodieHFileReader hoodieHFileReader = new HoodieHFileReader(context.getHadoopConf().get(), new Path(baseFile.getPath()),
+ new CacheConfig(context.getHadoopConf().get()));
+ List<Pair<String, IndexedRecord>> records = hoodieHFileReader.readAllRecords();
+ records.forEach(entry -> {
+ if (populateMetaFields) {
+ assertNotNull(((GenericRecord) entry.getSecond()).get(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ } else {
+ assertNull(((GenericRecord) entry.getSecond()).get(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ }
+ });
+ }
+
/**
* Test rollback of various table operations sync to Metadata Table correctly.
*/
@@ -586,6 +653,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
* Tests the metadata payload spurious deletes.
* Lets say a commit was applied to metadata table, and later was explicitly got rolledback. Due to spark task failures, there could be more files in rollback
* metadata when compared to the original commit metadata. When payload consistency check is enabled, it will throw exception. If not, it will succeed.
+ *
* @throws Exception
*/
@ParameterizedTest
@@ -1308,6 +1376,95 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
}
}
+ /**
+ * Fetching WriteConfig for metadata table from Data table's writeConfig is not trivial and the method is not public in source code. so, for now,
+ * using this method which mimics source code.
+ * @param writeConfig
+ * @return
+ */
+ private HoodieWriteConfig getMetadataWriteConfig(HoodieWriteConfig writeConfig) {
+ int parallelism = writeConfig.getMetadataInsertParallelism();
+
+ int minCommitsToKeep = Math.max(writeConfig.getMetadataMinCommitsToKeep(), writeConfig.getMinCommitsToKeep());
+ int maxCommitsToKeep = Math.max(writeConfig.getMetadataMaxCommitsToKeep(), writeConfig.getMaxCommitsToKeep());
+
+ // Create the write config for the metadata table by borrowing options from the main write config.
+ HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
+ .withTimelineLayoutVersion(TimelineLayoutVersion.CURR_VERSION)
+ .withConsistencyGuardConfig(ConsistencyGuardConfig.newBuilder()
+ .withConsistencyCheckEnabled(writeConfig.getConsistencyGuardConfig().isConsistencyCheckEnabled())
+ .withInitialConsistencyCheckIntervalMs(writeConfig.getConsistencyGuardConfig().getInitialConsistencyCheckIntervalMs())
+ .withMaxConsistencyCheckIntervalMs(writeConfig.getConsistencyGuardConfig().getMaxConsistencyCheckIntervalMs())
+ .withMaxConsistencyChecks(writeConfig.getConsistencyGuardConfig().getMaxConsistencyChecks())
+ .build())
+ .withWriteConcurrencyMode(WriteConcurrencyMode.SINGLE_WRITER)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).withFileListingParallelism(writeConfig.getFileListingParallelism()).build())
+ .withAutoCommit(true)
+ .withAvroSchemaValidate(true)
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ .withRollbackUsingMarkers(false)
+ .withPath(HoodieTableMetadata.getMetadataTableBasePath(writeConfig.getBasePath()))
+ .withSchema(HoodieMetadataRecord.getClassSchema().toString())
+ .forTable(writeConfig.getTableName() + METADATA_TABLE_NAME_SUFFIX)
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withAsyncClean(writeConfig.isMetadataAsyncClean())
+ // we will trigger cleaning manually, to control the instant times
+ .withAutoClean(false)
+ .withCleanerParallelism(parallelism)
+ .withCleanerPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS)
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY)
+ .retainCommits(writeConfig.getMetadataCleanerCommitsRetained())
+ .archiveCommitsWith(minCommitsToKeep, maxCommitsToKeep)
+ // we will trigger compaction manually, to control the instant times
+ .withInlineCompaction(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(writeConfig.getMetadataCompactDeltaCommitMax()).build())
+ .withParallelism(parallelism, parallelism)
+ .withDeleteParallelism(parallelism)
+ .withRollbackParallelism(parallelism)
+ .withFinalizeWriteParallelism(parallelism)
+ .withAllowMultiWriteOnSameInstant(true)
+ .withKeyGenerator(HoodieTableMetadataKeyGenerator.class.getCanonicalName())
+ .withPopulateMetaFields(writeConfig.getMetadataConfig().populateMetaFields());
+
+ // RecordKey properties are needed for the metadata table records
+ final Properties properties = new Properties();
+ properties.put(HoodieTableConfig.RECORDKEY_FIELDS.key(), HoodieMetadataPayload.SCHEMA_FIELD_ID_KEY);
+ properties.put("hoodie.datasource.write.recordkey.field", HoodieMetadataPayload.SCHEMA_FIELD_ID_KEY);
+ builder.withProperties(properties);
+
+ if (writeConfig.isMetricsOn()) {
+ builder.withMetricsConfig(HoodieMetricsConfig.newBuilder()
+ .withReporterType(writeConfig.getMetricsReporterType().toString())
+ .withExecutorMetrics(writeConfig.isExecutorMetricsEnabled())
+ .on(true).build());
+ switch (writeConfig.getMetricsReporterType()) {
+ case GRAPHITE:
+ builder.withMetricsGraphiteConfig(HoodieMetricsGraphiteConfig.newBuilder()
+ .onGraphitePort(writeConfig.getGraphiteServerPort())
+ .toGraphiteHost(writeConfig.getGraphiteServerHost())
+ .usePrefix(writeConfig.getGraphiteMetricPrefix()).build());
+ break;
+ case JMX:
+ builder.withMetricsJmxConfig(HoodieMetricsJmxConfig.newBuilder()
+ .onJmxPort(writeConfig.getJmxPort())
+ .toJmxHost(writeConfig.getJmxHost())
+ .build());
+ break;
+ case DATADOG:
+ case PROMETHEUS:
+ case PROMETHEUS_PUSHGATEWAY:
+ case CONSOLE:
+ case INMEMORY:
+ case CLOUDWATCH:
+ break;
+ default:
+ throw new HoodieMetadataException("Unsupported Metrics Reporter type " + writeConfig.getMetricsReporterType());
+ }
+ }
+ return builder.build();
+ }
+
private void doPreBootstrapOperations(HoodieTestTable testTable) throws Exception {
doPreBootstrapOperations(testTable, "0000001", "0000002");
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieMetadataBase.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieMetadataBase.java
index 0a45d31..25dfd29 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieMetadataBase.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieMetadataBase.java
@@ -292,7 +292,7 @@ public class TestHoodieMetadataBase extends HoodieClientTestHarness {
.enable(useFileListingMetadata)
.enableFullScan(enableFullScan)
.enableMetrics(enableMetrics)
- .withPopulateMetaFields(false)
+ .withPopulateMetaFields(HoodieMetadataConfig.POPULATE_META_FIELDS.defaultValue())
.ignoreSpuriousDeletes(validateMetadataPayloadConsistency)
.build())
.withMetricsConfig(HoodieMetricsConfig.newBuilder().on(enableMetrics)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
index 810e475..51791c9 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
@@ -118,23 +118,15 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.7.0")
.withDocumentation("Parallelism to use, when listing the table on lake storage.");
- public static final ConfigProperty<Boolean> ENABLE_INLINE_READING = ConfigProperty
- .key(METADATA_PREFIX + ".enable.inline.reading")
- .defaultValue(true)
- .sinceVersion("0.10.0")
- .withDocumentation("Enable inline reading of Log files. By default log block contents are read as byte[] using regular input stream and records "
- + "are deserialized from it. Enabling this will read each log block as an inline file and read records from the same. For instance, "
- + "for HFileDataBlock, a inline file will be read using HFileReader.");
-
public static final ConfigProperty<Boolean> ENABLE_FULL_SCAN_LOG_FILES = ConfigProperty
.key(METADATA_PREFIX + ".enable.full.scan.log.files")
.defaultValue(true)
.sinceVersion("0.10.0")
.withDocumentation("Enable full scanning of log files while reading log records. If disabled, hudi does look up of only interested entries.");
- public static final ConfigProperty<String> POPULATE_META_FIELDS = ConfigProperty
+ public static final ConfigProperty<Boolean> POPULATE_META_FIELDS = ConfigProperty
.key(METADATA_PREFIX + ".populate.meta.fields")
- .defaultValue("false")
+ .defaultValue(true)
.sinceVersion("0.10.0")
.withDocumentation("When enabled, populates all meta fields. When disabled, no meta fields are populated.");
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMergedLogRecordReader.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMergedLogRecordReader.java
index 2c9ca39..e635eea 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMergedLogRecordReader.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMergedLogRecordReader.java
@@ -28,6 +28,8 @@ import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
import org.apache.hadoop.fs.FileSystem;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.util.SpillableMapUtils;
import org.apache.log4j.LogManager;
@@ -142,7 +144,7 @@ public class HoodieMetadataMergedLogRecordReader extends HoodieMergedLogRecordSc
*/
public static class Builder extends HoodieMergedLogRecordScanner.Builder {
private Set<String> mergeKeyFilter = Collections.emptySet();
- private boolean enableFullScan;
+ private boolean enableFullScan = HoodieMetadataConfig.ENABLE_FULL_SCAN_LOG_FILES.defaultValue();
private boolean enableInlineReading;
@Override
diff --git a/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.COPY_ON_WRITE.zip b/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.COPY_ON_WRITE.zip
index 42c0301..299b070 100644
Binary files a/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.COPY_ON_WRITE.zip and b/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.COPY_ON_WRITE.zip differ
diff --git a/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.MERGE_ON_READ.zip b/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.MERGE_ON_READ.zip
index 8e57b1d..d80439d 100644
Binary files a/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.MERGE_ON_READ.zip and b/hudi-utilities/src/test/resources/fixtures/testUpsertsContinuousModeWithMultipleWriters.MERGE_ON_READ.zip differ
[hudi] 06/26: [HUDI-2880] Fixing loading of props from default dir (#4167)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 9d7f681ca6d3d8c244854040298569f54ba701b8
Author: Sivabalan Narayanan <si...@uber.com>
AuthorDate: Wed Dec 1 03:02:30 2021 -0500
[HUDI-2880] Fixing loading of props from default dir (#4167)
* Fixing loading of props from default dir
* addressing comments
(cherry picked from commit f4c25ba3fd11e2da1adcf2e089969a0b7f3dc62e)
---
.../common/config/DFSPropertiesConfiguration.java | 31 ++++++++++++----------
.../util/TestDFSPropertiesConfiguration.java | 5 ++--
2 files changed, 19 insertions(+), 17 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java b/hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
index 01585f9..afce8f6 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
@@ -18,18 +18,21 @@
package org.apache.hudi.common.config;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
@@ -53,10 +56,9 @@ public class DFSPropertiesConfiguration {
private static final Logger LOG = LogManager.getLogger(DFSPropertiesConfiguration.class);
public static final String DEFAULT_PROPERTIES_FILE = "hudi-defaults.conf";
-
public static final String CONF_FILE_DIR_ENV_NAME = "HUDI_CONF_DIR";
-
public static final String DEFAULT_CONF_FILE_DIR = "file:/etc/hudi/conf";
+ public static final Path DEFAULT_PATH = new Path(DEFAULT_CONF_FILE_DIR + "/" + DEFAULT_PROPERTIES_FILE);
// props read from hudi-defaults.conf
private static TypedProperties GLOBAL_PROPS = loadGlobalProps();
@@ -97,11 +99,7 @@ public class DFSPropertiesConfiguration {
if (defaultConfPath.isPresent()) {
conf.addPropsFromFile(defaultConfPath.get());
} else {
- try {
- conf.addPropsFromFile(new Path(DEFAULT_CONF_FILE_DIR));
- } catch (Exception ignored) {
- LOG.warn("Didn't find config file under default conf file dir: " + DEFAULT_CONF_FILE_DIR);
- }
+ conf.addPropsFromFile(DEFAULT_PATH);
}
return conf.getProps();
}
@@ -128,14 +126,19 @@ public class DFSPropertiesConfiguration {
filePath.toString(),
Option.ofNullable(hadoopConfig).orElseGet(Configuration::new)
);
+ try {
+ if (filePath.equals(DEFAULT_PATH) && !fs.exists(filePath)) {
+ LOG.warn("Properties file " + filePath + " not found. Ignoring to load props file");
+ return;
+ }
- try (BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(filePath)))) {
+ BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(filePath)));
visitedFilePaths.add(filePath.toString());
currentFilePath = filePath;
addPropsFromStream(reader);
} catch (IOException ioe) {
- LOG.error("Error reading in properties from dfs");
- throw new IllegalArgumentException("Cannot read properties from dfs", ioe);
+ LOG.error("Error reading in properties from dfs from file " + filePath);
+ throw new HoodieIOException("Cannot read properties from dfs from file " + filePath, ioe);
}
}
@@ -192,7 +195,7 @@ public class DFSPropertiesConfiguration {
}
private String[] splitProperty(String line) {
- line = line.replaceAll("\\s+"," ");
+ line = line.replaceAll("\\s+", " ");
String delimiter = line.contains("=") ? "=" : " ";
int ind = line.indexOf(delimiter);
String k = line.substring(0, ind).trim();
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java b/hudi-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
index 6a7075f..4657393 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
@@ -145,7 +145,7 @@ public class TestDFSPropertiesConfiguration {
}
@Test
- public void testLocalFileSystemLoading() {
+ public void testLocalFileSystemLoading() throws IOException {
DFSPropertiesConfiguration cfg = new DFSPropertiesConfiguration(dfs.getConf(), new Path(dfsBasePath + "/t1.props"));
cfg.addPropsFromFile(
@@ -156,8 +156,7 @@ public class TestDFSPropertiesConfiguration {
.getResource("props/test.properties")
.getPath()
)
- )
- );
+ ));
TypedProperties props = cfg.getProps();
[hudi] 08/26: Fixed partitions produced by layout optimization in case order-by key is composed of a single column (#4183)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 1ee35c484b9e4eefb033086c5e0388afac70e642
Author: Alexey Kudinkin <al...@infinilake.com>
AuthorDate: Wed Dec 1 20:56:04 2021 -0800
Fixed partitions produced by layout optimization in case order-by key is composed of a single column (#4183)
(cherry picked from commit 772f5ca24ee76c5c3676a9136fd994f7cd55a0e3)
---
.../src/main/java/org/apache/spark/OrderingIndexHelper.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/spark/OrderingIndexHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/spark/OrderingIndexHelper.java
index 67b1c67..3d0138e 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/spark/OrderingIndexHelper.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/spark/OrderingIndexHelper.java
@@ -100,7 +100,7 @@ public class OrderingIndexHelper {
}
// only one col to sort, no need to use z-order
if (sortCols.size() == 1) {
- return df.repartitionByRange(fieldNum, org.apache.spark.sql.functions.col(sortCols.get(0)));
+ return df.repartitionByRange(fileNum, org.apache.spark.sql.functions.col(sortCols.get(0)));
}
Map<Integer, StructField> fieldMap = sortCols
.stream().collect(Collectors.toMap(e -> Arrays.asList(df.schema().fields()).indexOf(columnsMap.get(e)), e -> columnsMap.get(e)));
[hudi] 20/26: [MINOR] Mitigate CI jobs timeout issues (#4173)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-0.10.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 31947a09a699f59578fe99c5834e171ab5f82280
Author: Raymond Xu <27...@users.noreply.github.com>
AuthorDate: Fri Dec 3 21:08:32 2021 -0800
[MINOR] Mitigate CI jobs timeout issues (#4173)
* skip shutdown zookeeper in `@AfterAll` in TestHBaseIndex
* rebalance CI tests
(cherry picked from commit a799fae31610b2c38ca5aa52c87fa92363a41a70)
---
azure-pipelines.yml | 14 +++++++-------
.../org/apache/hudi/client/functional/TestHBaseIndex.java | 6 +++++-
2 files changed, 12 insertions(+), 8 deletions(-)
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index ef68cd5..2249328 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -99,7 +99,7 @@ stages:
jdkVersionOption: '1.8'
mavenOptions: '-Xmx2g $(MAVEN_OPTS)'
- job: UT_FT_3
- displayName: UT FT cli & utilities & sync/hive-sync
+ displayName: UT FT clients & cli & utilities & sync/hive-sync
steps:
- task: Cache@2
displayName: set cache
@@ -119,20 +119,20 @@ stages:
jdkVersionOption: '1.8'
mavenOptions: '-Xmx2g $(MAVEN_OPTS)'
- task: Maven@3
- displayName: UT cli & utilities & sync/hive-sync
+ displayName: UT clients & cli & utilities & sync/hive-sync
inputs:
mavenPomFile: 'pom.xml'
goals: 'test'
- options: -Punit-tests -pl hudi-cli,hudi-utilities,hudi-sync/hudi-hive-sync
+ options: -Punit-tests -pl hudi-client/hudi-client-common,hudi-client/hudi-flink-client,hudi-client/hudi-java-client,hudi-cli,hudi-utilities,hudi-sync/hudi-hive-sync
publishJUnitResults: false
jdkVersionOption: '1.8'
mavenOptions: '-Xmx2g $(MAVEN_OPTS)'
- task: Maven@3
- displayName: FT cli & utilities & sync/hive-sync
+ displayName: FT clients & cli & utilities & sync/hive-sync
inputs:
mavenPomFile: 'pom.xml'
goals: 'test'
- options: -Pfunctional-tests -pl hudi-cli,hudi-utilities,hudi-sync/hudi-hive-sync
+ options: -Pfunctional-tests -pl hudi-client/hudi-client-common,hudi-client/hudi-flink-client,hudi-client/hudi-java-client,hudi-cli,hudi-utilities,hudi-sync/hudi-hive-sync
publishJUnitResults: false
jdkVersionOption: '1.8'
mavenOptions: '-Xmx2g $(MAVEN_OPTS)'
@@ -161,7 +161,7 @@ stages:
inputs:
mavenPomFile: 'pom.xml'
goals: 'test'
- options: -Punit-tests -pl !hudi-common,!hudi-flink,!hudi-client/hudi-spark-client,!hudi-cli,!hudi-utilities,!hudi-sync/hudi-hive-sync
+ options: -Punit-tests -pl !hudi-common,!hudi-flink,!hudi-client/hudi-spark-client,!hudi-client/hudi-client-common,!hudi-client/hudi-flink-client,!hudi-client/hudi-java-client,!hudi-cli,!hudi-utilities,!hudi-sync/hudi-hive-sync
publishJUnitResults: false
jdkVersionOption: '1.8'
mavenOptions: '-Xmx2g $(MAVEN_OPTS)'
@@ -170,7 +170,7 @@ stages:
inputs:
mavenPomFile: 'pom.xml'
goals: 'test'
- options: -Pfunctional-tests -pl !hudi-common,!hudi-flink,!hudi-client/hudi-spark-client,!hudi-cli,!hudi-utilities,!hudi-sync/hudi-hive-sync
+ options: -Pfunctional-tests -pl !hudi-common,!hudi-flink,!hudi-client/hudi-spark-client,!hudi-client/hudi-client-common,!hudi-client/hudi-flink-client,!hudi-client/hudi-java-client,!hudi-cli,!hudi-utilities,!hudi-sync/hudi-hive-sync
publishJUnitResults: false
jdkVersionOption: '1.8'
mavenOptions: '-Xmx2g $(MAVEN_OPTS)'
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHBaseIndex.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHBaseIndex.java
index 43bf36c..b35fee0 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHBaseIndex.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHBaseIndex.java
@@ -119,7 +119,11 @@ public class TestHBaseIndex extends SparkClientFunctionalTestHarness {
@AfterAll
public static void clean() throws Exception {
- utility.shutdownMiniCluster();
+ utility.shutdownMiniHBaseCluster();
+ utility.shutdownMiniDFSCluster();
+ utility.shutdownMiniMapReduceCluster();
+ // skip shutdownZkCluster due to localhost connection refused issue
+ utility = null;
}
@BeforeEach