You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/05/14 21:16:28 UTC
(hudi) branch branch-0.x updated (3d5d274847c -> 3f78130d007)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a change to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
from 3d5d274847c [HUDI-7473] Rebalance CI (#10805)
new 45923f3cacd [HUDI-6947] Refactored HoodieSchemaUtils.deduceWriterSchema with many flags (#10810)
new 695577bdc95 [HUDI-7356] Passing configs to file reader constructor for flexibility (#10698)
new 4680cb45333 [HUDI-7197] Adding mis fixes related with table services testing (#10280)
new 9f00f6d6ed7 [HUDI-5167] Reducing total test run time: reducing tests for virtual keys (#7153)
new 8ed8a20b690 [HUDI-7488] The BigQuerySyncTool can't work well when the hudi table schema changed (#10830)
new 06584c6f768 [MINOR] Separate HoodieSparkWriterTestBase to reduce duplication (#10832)
new 8e6eff945bf [HUDI-7491] Fixing handling null values of extra metadata in clean commit metadata (#10837)
new dbe16f3f965 [HUDI-7411] Meta sync should consider cleaner commit (#10676)
new 866348adb39 [ENG-6316] Bump cleaner retention for MDT (#537) (#10655)
new 632e61ff2d6 [HUDI-6043] Metadata Table should use default values for Compaction preserveCommitMetadata field (#8393)
new 02ae11f4b46 [HUDI-5101] Adding spark-structured streaming test support via spark-submit job (#7074)
new 45a2e071a4e [HUDI-7495] Bump mysql-connector-java from 8.0.22 to 8.0.28 in /hudi-platform-service/hudi-metaserver/hudi-metaserver-server (#7674)
new 80990d42325 [HUDI-7163] Fix not parsable text DateTimeParseException when compact (#10220)
new 3f78130d007 [HUDI-7496] Bump mybatis from 3.4.6 to 3.5.6 in /hudi-platform-service/hudi-metaserver/hudi-metaserver-server (#7673)
The 14 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:
.../org/apache/hudi/io/HoodieCreateHandle.java | 4 +-
.../java/org/apache/hudi/io/HoodieMergeHandle.java | 4 +-
.../hudi/metadata/HoodieMetadataWriteUtils.java | 28 +++-
.../table/action/clean/CleanActionExecutor.java | 2 +-
.../hudi/table/action/clean/CleanPlanner.java | 4 +-
.../action/commit/BaseCommitActionExecutor.java | 1 +
.../compact/ScheduleCompactionActionExecutor.java | 9 +-
.../metadata/TestHoodieMetadataWriteUtils.java | 64 ++++++++
.../apache/hudi/table/action/TestCleanPlanner.java | 2 +-
.../io/storage/HoodieSparkFileReaderFactory.java | 9 +-
.../TestHoodieClientOnCopyOnWriteStorage.java | 112 ++++++-------
.../hudi/common/table/timeline/TimelineUtils.java | 27 +++-
.../io/storage/HoodieAvroFileReaderFactory.java | 13 +-
.../hudi/io/storage/HoodieFileReaderFactory.java | 20 +--
.../hudi/common/table/TestTimelineUtils.java | 46 +++++-
.../java/org/apache/hudi/util/StreamerUtil.java | 39 ++---
.../gcp/bigquery/HoodieBigQuerySyncClient.java | 25 +--
.../streaming/StructuredStreamingSinkUtil.java | 168 ++++++++++++++++++++
.../StructuredStreamingSinkTestWriter.scala | 104 ++++++++++++
.../hudi-metaserver/hudi-metaserver-server/pom.xml | 4 +-
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 176 +++++++++++----------
.../sql/catalyst/catalog/HoodieCatalogTable.scala | 7 +-
.../apache/hudi/HoodieSparkWriterTestBase.scala | 136 ++++++++++++++++
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 152 +++---------------
.../apache/hudi/TestHoodieSparkSqlWriterUtc.scala | 2 +-
.../hudi/TestTableSchemaResolverWithSparkSQL.scala | 102 +-----------
.../apache/hudi/sync/common/HoodieSyncClient.java | 5 +-
.../org/apache/hudi/utilities/UtilHelpers.java | 1 +
.../apache/hudi/utilities/TestHoodieIndexer.java | 13 +-
29 files changed, 809 insertions(+), 470 deletions(-)
create mode 100644 hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
create mode 100644 hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkUtil.java
create mode 100644 hudi-integ-test/src/main/scala/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkTestWriter.scala
create mode 100644 hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/HoodieSparkWriterTestBase.scala
(hudi) 05/14: [HUDI-7488] The BigQuerySyncTool can't work well when the hudi table schema changed (#10830)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 8ed8a20b6902b95a3e698a519822bdb0210a0a7f
Author: steve-xi-awx <84...@users.noreply.github.com>
AuthorDate: Fri Mar 8 08:29:03 2024 +0800
[HUDI-7488] The BigQuerySyncTool can't work well when the hudi table schema changed (#10830)
---
.../gcp/bigquery/HoodieBigQuerySyncClient.java | 25 ++++++++++++++--------
1 file changed, 16 insertions(+), 9 deletions(-)
diff --git a/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/HoodieBigQuerySyncClient.java b/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/HoodieBigQuerySyncClient.java
index 5a23a4079ae..32430b53329 100644
--- a/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/HoodieBigQuerySyncClient.java
+++ b/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/HoodieBigQuerySyncClient.java
@@ -44,6 +44,7 @@ import com.google.cloud.bigquery.Table;
import com.google.cloud.bigquery.TableId;
import com.google.cloud.bigquery.TableInfo;
import com.google.cloud.bigquery.ViewDefinition;
+import com.google.cloud.bigquery.StandardTableDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -198,16 +199,22 @@ public class HoodieBigQuerySyncClient extends HoodieSyncClient {
LOG.info("No table update is needed.");
return; // No need to update schema.
}
- ExternalTableDefinition.Builder builder = definition.toBuilder();
- builder.setSchema(finalSchema);
- builder.setAutodetect(false);
- if (definition.getHivePartitioningOptions() != null) {
- builder.setHivePartitioningOptions(definition.getHivePartitioningOptions().toBuilder().setRequirePartitionFilter(requirePartitionFilter).build());
+ if (!StringUtils.isNullOrEmpty(bigLakeConnectionId)) {
+ Table updatedTable =
+ existingTable.toBuilder().setDefinition(StandardTableDefinition.of(finalSchema)).build();
+ updatedTable.update();
+ } else {
+ ExternalTableDefinition.Builder builder = definition.toBuilder();
+ builder.setSchema(finalSchema);
+ builder.setAutodetect(false);
+ if (definition.getHivePartitioningOptions() != null) {
+ builder.setHivePartitioningOptions(definition.getHivePartitioningOptions().toBuilder().setRequirePartitionFilter(requirePartitionFilter).build());
+ }
+ Table updatedTable = existingTable.toBuilder()
+ .setDefinition(builder.build())
+ .build();
+ bigquery.update(updatedTable);
}
- Table updatedTable = existingTable.toBuilder()
- .setDefinition(builder.build())
- .build();
- bigquery.update(updatedTable);
}
public void createVersionsTable(String tableName, String sourceUri, String sourceUriPrefix, List<String> partitionFields) {
(hudi) 08/14: [HUDI-7411] Meta sync should consider cleaner commit (#10676)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit dbe16f3f965a76c94804e19063df0253dd6e69d7
Author: Sagar Sumit <sa...@gmail.com>
AuthorDate: Fri Mar 8 23:34:53 2024 +0530
[HUDI-7411] Meta sync should consider cleaner commit (#10676)
---
.../hudi/common/table/timeline/TimelineUtils.java | 27 ++++++++++---
.../hudi/common/table/TestTimelineUtils.java | 46 ++++++++++++++++++++--
.../sql/catalyst/catalog/HoodieCatalogTable.scala | 7 +---
.../apache/hudi/sync/common/HoodieSyncClient.java | 5 +--
4 files changed, 67 insertions(+), 18 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
index 52788acc437..ca6d5b57907 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
@@ -81,13 +81,15 @@ public class TimelineUtils {
}
/**
- * Returns partitions that have been deleted or marked for deletion in the given timeline.
+ * Returns partitions that have been deleted or marked for deletion in the timeline between given commit time range.
* Does not include internal operations such as clean in the timeline.
*/
- public static List<String> getDroppedPartitions(HoodieTimeline timeline) {
+ public static List<String> getDroppedPartitions(HoodieTableMetaClient metaClient, Option<String> lastCommitTimeSynced, Option<String> lastCommitCompletionTimeSynced) {
+ HoodieTimeline timeline = lastCommitTimeSynced.isPresent()
+ ? TimelineUtils.getCommitsTimelineAfter(metaClient, lastCommitTimeSynced.get(), lastCommitCompletionTimeSynced)
+ : metaClient.getActiveTimeline();
HoodieTimeline completedTimeline = timeline.getWriteTimeline().filterCompletedInstants();
HoodieTimeline replaceCommitTimeline = completedTimeline.getCompletedReplaceTimeline();
-
Map<String, String> partitionToLatestDeleteTimestamp = replaceCommitTimeline.getInstantsAsStream()
.map(instant -> {
try {
@@ -102,6 +104,21 @@ public class TimelineUtils {
.flatMap(pair -> pair.getRight().getPartitionToReplaceFileIds().keySet().stream()
.map(partition -> new AbstractMap.SimpleEntry<>(partition, pair.getLeft().getTimestamp()))
).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (existing, replace) -> replace));
+ // cleaner could delete a partition when there are no active filegroups in the partition
+ HoodieTimeline cleanerTimeline = metaClient.getActiveTimeline().getCleanerTimeline().filterCompletedInstants();
+ cleanerTimeline.getInstantsAsStream()
+ .forEach(instant -> {
+ try {
+ HoodieCleanMetadata cleanMetadata = TimelineMetadataUtils.deserializeHoodieCleanMetadata(cleanerTimeline.getInstantDetails(instant).get());
+ cleanMetadata.getPartitionMetadata().forEach((partition, partitionMetadata) -> {
+ if (partitionMetadata.getIsPartitionDeleted()) {
+ partitionToLatestDeleteTimestamp.put(partition, instant.getTimestamp());
+ }
+ });
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to get partitions cleaned at " + instant, e);
+ }
+ });
if (partitionToLatestDeleteTimestamp.isEmpty()) {
// There is no dropped partitions
@@ -244,7 +261,7 @@ public class TimelineUtils {
return false;
} catch (IOException e) {
- throw new HoodieIOException("Unable to read instant information: " + instant + " for " + metaClient.getBasePath(), e);
+ throw new HoodieIOException("Unable to read instant information: " + instant + " for " + metaClient.getBasePathV2().toString(), e);
}
}
@@ -440,7 +457,7 @@ public class TimelineUtils {
}
public enum HollowCommitHandling {
- FAIL, BLOCK, USE_TRANSITION_TIME;
+ FAIL, BLOCK, USE_TRANSITION_TIME
}
public static boolean isDeletePartition(WriteOperationType operation) {
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
index 842366940da..eef515c6ada 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
@@ -158,7 +158,7 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
HoodieInstant cleanInstant = new HoodieInstant(true, CLEAN_ACTION, ts);
activeTimeline.createNewInstant(cleanInstant);
- activeTimeline.saveAsComplete(cleanInstant, getCleanMetadata(olderPartition, ts));
+ activeTimeline.saveAsComplete(cleanInstant, getCleanMetadata(olderPartition, ts, false));
}
metaClient.reloadActiveTimeline();
@@ -197,7 +197,7 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
HoodieInstant cleanInstant = new HoodieInstant(true, CLEAN_ACTION, ts);
activeTimeline.createNewInstant(cleanInstant);
- activeTimeline.saveAsComplete(cleanInstant, getCleanMetadata(partitionPath, ts));
+ activeTimeline.saveAsComplete(cleanInstant, getCleanMetadata(partitionPath, ts, false));
}
metaClient.reloadActiveTimeline();
@@ -553,7 +553,7 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
return getUTF8Bytes(commit.toJsonString());
}
- private Option<byte[]> getCleanMetadata(String partition, String time) throws IOException {
+ private Option<byte[]> getCleanMetadata(String partition, String time, boolean isPartitionDeleted) throws IOException {
Map<String, HoodieCleanPartitionMetadata> partitionToFilesCleaned = new HashMap<>();
List<String> filesDeleted = new ArrayList<>();
filesDeleted.add("file-" + partition + "-" + time + "1");
@@ -564,6 +564,7 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
.setFailedDeleteFiles(Collections.emptyList())
.setDeletePathPatterns(Collections.emptyList())
.setSuccessDeleteFiles(filesDeleted)
+ .setIsPartitionDeleted(isPartitionDeleted)
.build();
partitionToFilesCleaned.putIfAbsent(partition, partitionMetadata);
HoodieCleanMetadata cleanMetadata = HoodieCleanMetadata.newBuilder()
@@ -611,4 +612,43 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
fail("should cover all handling mode.");
}
}
+
+ @Test
+ public void testGetDroppedPartitions() throws Exception {
+ HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
+ HoodieTimeline activeCommitTimeline = activeTimeline.getCommitTimeline();
+ assertTrue(activeCommitTimeline.empty());
+
+ String olderPartition = "p1"; // older partitions that will be deleted by clean commit
+ // first insert to the older partition
+ HoodieInstant instant1 = new HoodieInstant(true, COMMIT_ACTION, "00001");
+ activeTimeline.createNewInstant(instant1);
+ activeTimeline.saveAsComplete(instant1, Option.of(getCommitMetadata(basePath, olderPartition, "00001", 2, Collections.emptyMap())));
+
+ metaClient.reloadActiveTimeline();
+ List<String> droppedPartitions = TimelineUtils.getDroppedPartitions(metaClient, Option.empty(), Option.empty());
+ // no dropped partitions
+ assertEquals(0, droppedPartitions.size());
+
+ // another commit inserts to new partition
+ HoodieInstant instant2 = new HoodieInstant(true, COMMIT_ACTION, "00002");
+ activeTimeline.createNewInstant(instant2);
+ activeTimeline.saveAsComplete(instant2, Option.of(getCommitMetadata(basePath, "p2", "00002", 2, Collections.emptyMap())));
+
+ metaClient.reloadActiveTimeline();
+ droppedPartitions = TimelineUtils.getDroppedPartitions(metaClient, Option.empty(), Option.empty());
+ // no dropped partitions
+ assertEquals(0, droppedPartitions.size());
+
+ // clean commit deletes older partition
+ HoodieInstant cleanInstant = new HoodieInstant(true, CLEAN_ACTION, "00003");
+ activeTimeline.createNewInstant(cleanInstant);
+ activeTimeline.saveAsComplete(cleanInstant, getCleanMetadata(olderPartition, "00003", true));
+
+ metaClient.reloadActiveTimeline();
+ droppedPartitions = TimelineUtils.getDroppedPartitions(metaClient, Option.empty(), Option.empty());
+ // older partition is in the list dropped partitions
+ assertEquals(1, droppedPartitions.size());
+ assertEquals(olderPartition, droppedPartitions.get(0));
+ }
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala
index 5fcc750ac5b..b194be57f7a 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/HoodieCatalogTable.scala
@@ -159,11 +159,6 @@ class HoodieCatalogTable(val spark: SparkSession, var table: CatalogTable) exten
StructType(tableSchema.filterNot(f => partitionFields.contains(f.name)))
}
- /**
- * The schema of data fields not including hoodie meta fields
- */
- lazy val dataSchemaWithoutMetaFields: StructType = removeMetaFields(dataSchema)
-
/**
* The schema of partition fields
*/
@@ -173,7 +168,7 @@ class HoodieCatalogTable(val spark: SparkSession, var table: CatalogTable) exten
* All the partition paths, excludes lazily deleted partitions.
*/
def getPartitionPaths: Seq[String] = {
- val droppedPartitions = TimelineUtils.getDroppedPartitions(metaClient.getActiveTimeline)
+ val droppedPartitions = TimelineUtils.getDroppedPartitions(metaClient, org.apache.hudi.common.util.Option.empty(), org.apache.hudi.common.util.Option.empty())
getAllPartitionPaths(spark, table)
.filter(!droppedPartitions.contains(_))
diff --git a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java
index 2c2d77651cb..9078e9d0711 100644
--- a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java
+++ b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java
@@ -93,10 +93,7 @@ public abstract class HoodieSyncClient implements HoodieMetaSyncOperations, Auto
* Going through archive timeline is a costly operation, and it should be avoided unless some start time is given.
*/
public Set<String> getDroppedPartitionsSince(Option<String> lastCommitTimeSynced, Option<String> lastCommitCompletionTimeSynced) {
- HoodieTimeline timeline = lastCommitTimeSynced.isPresent()
- ? TimelineUtils.getCommitsTimelineAfter(metaClient, lastCommitTimeSynced.get(), lastCommitCompletionTimeSynced)
- : metaClient.getActiveTimeline();
- return new HashSet<>(TimelineUtils.getDroppedPartitions(timeline));
+ return new HashSet<>(TimelineUtils.getDroppedPartitions(metaClient, lastCommitTimeSynced, lastCommitCompletionTimeSynced));
}
@Override
(hudi) 10/14: [HUDI-6043] Metadata Table should use default values for Compaction preserveCommitMetadata field (#8393)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 632e61ff2d60bcaf158c018dd6919ea29d57be6f
Author: Lokesh Jain <lj...@apache.org>
AuthorDate: Sat Mar 9 22:09:50 2024 +0530
[HUDI-6043] Metadata Table should use default values for Compaction preserveCommitMetadata field (#8393)
---
.../src/main/java/org/apache/hudi/io/HoodieCreateHandle.java | 4 +---
.../src/main/java/org/apache/hudi/io/HoodieMergeHandle.java | 4 +---
2 files changed, 2 insertions(+), 6 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
index bdb35641f26..0a0f3352069 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
@@ -34,7 +34,6 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieInsertException;
import org.apache.hudi.io.storage.HoodieFileWriter;
import org.apache.hudi.io.storage.HoodieFileWriterFactory;
-import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.table.HoodieTable;
import org.apache.avro.Schema;
@@ -115,8 +114,7 @@ public class HoodieCreateHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O
public HoodieCreateHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
String partitionPath, String fileId, Map<String, HoodieRecord<T>> recordMap,
TaskContextSupplier taskContextSupplier) {
- // preserveMetadata is disabled by default for MDT but enabled otherwise
- this(config, instantTime, hoodieTable, partitionPath, fileId, taskContextSupplier, !HoodieTableMetadata.isMetadataTable(config.getBasePath()));
+ this(config, instantTime, hoodieTable, partitionPath, fileId, taskContextSupplier, true);
this.recordMap = recordMap;
this.useWriterSchema = true;
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
index 4460e29c8a4..b6d13164f37 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
@@ -47,7 +47,6 @@ import org.apache.hudi.io.storage.HoodieFileReaderFactory;
import org.apache.hudi.io.storage.HoodieFileWriter;
import org.apache.hudi.io.storage.HoodieFileWriterFactory;
import org.apache.hudi.keygen.BaseKeyGenerator;
-import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.table.HoodieTable;
import org.apache.avro.Schema;
@@ -144,8 +143,7 @@ public class HoodieMergeHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O>
super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier);
this.keyToNewRecords = keyToNewRecords;
this.useWriterSchemaForCompaction = true;
- // preserveMetadata is disabled by default for MDT but enabled otherwise
- this.preserveMetadata = !HoodieTableMetadata.isMetadataTable(config.getBasePath());
+ this.preserveMetadata = true;
init(fileId, this.partitionPath, dataFileToBeMerged);
validateAndSetAndKeyGenProps(keyGeneratorOpt, config.populateMetaFields());
}
(hudi) 12/14: [HUDI-7495] Bump mysql-connector-java from 8.0.22 to 8.0.28 in /hudi-platform-service/hudi-metaserver/hudi-metaserver-server (#7674)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 45a2e071a4eccec1cc39cfce8758b3384dac8caa
Author: dependabot[bot] <49...@users.noreply.github.com>
AuthorDate: Sat Mar 9 10:51:44 2024 -0800
[HUDI-7495] Bump mysql-connector-java from 8.0.22 to 8.0.28 in /hudi-platform-service/hudi-metaserver/hudi-metaserver-server (#7674)
Signed-off-by: dependabot[bot] <su...@github.com>
Co-authored-by: dependabot[bot] <49...@users.noreply.github.com>
---
hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml b/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml
index 10ac5be853a..8b32f962d7c 100644
--- a/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml
+++ b/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml
@@ -32,7 +32,7 @@
<main.basedir>${project.parent.basedir}</main.basedir>
<mybatis.version>3.4.6</mybatis.version>
<HikariCP.version>4.0.3</HikariCP.version>
- <mysql-connector-java.version>8.0.22</mysql-connector-java.version>
+ <mysql-connector-java.version>8.0.28</mysql-connector-java.version>
</properties>
<dependencies>
(hudi) 11/14: [HUDI-5101] Adding spark-structured streaming test support via spark-submit job (#7074)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 02ae11f4b46255af43a4b60b9e45fe3059c84408
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Sat Mar 9 09:33:10 2024 -0800
[HUDI-5101] Adding spark-structured streaming test support via spark-submit job (#7074)
Co-authored-by: Y Ethan Guo <et...@gmail.com>
---
.../streaming/StructuredStreamingSinkUtil.java | 168 +++++++++++++++++++++
.../StructuredStreamingSinkTestWriter.scala | 104 +++++++++++++
2 files changed, 272 insertions(+)
diff --git a/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkUtil.java b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkUtil.java
new file mode 100644
index 00000000000..f6fec62cb3b
--- /dev/null
+++ b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkUtil.java
@@ -0,0 +1,168 @@
+/*
+ * 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.integ.testsuite.streaming;
+
+import org.apache.hudi.exception.HoodieException;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Saprk-submit to test spark streaming
+ *
+ * Sample command.
+ * ./bin/spark-submit --master local[2] --driver-memory 1g --executor-memory 1g \
+ * --class org.apache.hudi.streaming.StructuredStreamingSinkUtil PATH TO hudi-integ-test-bundle-0.13.0-SNAPSHOT.jar \
+ * --spark-master local[2] \
+ * --source-path /tmp/parquet_ny/ \
+ * --target-path /tmp/hudi_streaming_kafka10/MERGE_ON_READ3/ \
+ * --checkpoint-path /tmp/hudi_streaming_kafka10/checkpoint_mor3/ \
+ * --table-type COPY_ON_WRITE \
+ * --partition-field date_col \
+ * --record-key-field tpep_pickup_datetime \
+ * --pre-combine-field tpep_dropoff_datetime \
+ * --table-name test_tbl
+ *
+ * Ensure "source-path" has parquet data.
+ */
+public class StructuredStreamingSinkUtil implements Serializable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StructuredStreamingSinkUtil.class);
+
+ private transient JavaSparkContext jsc;
+ private SparkSession sparkSession;
+ private Config cfg;
+
+ public StructuredStreamingSinkUtil(JavaSparkContext jsc, Config cfg) {
+ this.jsc = jsc;
+ this.sparkSession = SparkSession.builder().config(jsc.getConf()).getOrCreate();
+ this.cfg = cfg;
+ }
+
+ public static class Config implements Serializable {
+ @Parameter(names = {"--source-path", "-sp"}, description = "Source path to consume data from", required = true)
+ public String sourcePath = null;
+
+ @Parameter(names = {"--target-path", "-tp"}, description = "Target path of the table of interest.", required = true)
+ public String targetPath = null;
+
+ @Parameter(names = {"--table-type", "-ty"}, description = "Target path of the table of interest.", required = true)
+ public String tableType = "COPY_ON_WRITE";
+
+ @Parameter(names = {"--checkpoint-path", "-cp"}, description = "Checkppint path of the table of interest", required = true)
+ public String checkpointPath = null;
+
+ @Parameter(names = {"--partition-field", "-pp"}, description = "Partitioning field", required = true)
+ public String partitionField = null;
+
+ @Parameter(names = {"--record-key-field", "-rk"}, description = "record key field", required = true)
+ public String recordKeyField = null;
+
+ @Parameter(names = {"--pre-combine-field", "-pc"}, description = "Precombine field", required = true)
+ public String preCombineField = null;
+
+ @Parameter(names = {"--table-name", "-tn"}, description = "Table name", required = true)
+ public String tableName = null;
+
+ @Parameter(names = {"--disable-metadata", "-dmdt"}, description = "Disable metadata while querying", required = false)
+ public Boolean disableMetadata = false;
+
+ @Parameter(names = {"--spark-master", "-ms"}, description = "Spark master", required = false)
+ public String sparkMaster = null;
+
+ @Parameter(names = {"--spark-memory", "-sm"}, description = "spark memory to use", required = false)
+ public String sparkMemory = "1g";
+
+ @Parameter(names = {"--help", "-h"}, help = true)
+ public Boolean help = false;
+
+ }
+
+ public static void main(String[] args) {
+ final Config cfg = new Config();
+ JCommander cmd = new JCommander(cfg, null, args);
+
+ if (cfg.help || args.length == 0) {
+ cmd.usage();
+ System.exit(1);
+ }
+
+ SparkConf sparkConf = buildSparkConf("Spark-structured-streaming-test", cfg.sparkMaster);
+ sparkConf.set("spark.executor.memory", cfg.sparkMemory);
+ JavaSparkContext jsc = new JavaSparkContext(sparkConf);
+
+ try {
+ StructuredStreamingSinkUtil streamingSinkUtil = new StructuredStreamingSinkUtil(jsc, cfg);
+ streamingSinkUtil.run();
+ } catch (Throwable throwable) {
+ LOG.error("Fail to execute tpcds read benchmarks for " + cfg, throwable);
+ } finally {
+ jsc.stop();
+ }
+ }
+
+ public void run() {
+ try {
+ LOG.info(cfg.toString());
+ StructuredStreamingSinkTestWriter.triggerStreaming(sparkSession, cfg.tableType, cfg.sourcePath, cfg.targetPath, cfg.checkpointPath,
+ cfg.tableName, cfg.partitionField, cfg.recordKeyField, cfg.preCombineField);
+ StructuredStreamingSinkTestWriter.waitUntilCondition(1000 * 60 * 10, 1000 * 30);
+ } catch (Exception e) {
+ throw new HoodieException("Unable to test spark structured writes to hudi " + cfg.targetPath, e);
+ } finally {
+ LOG.warn("Completing Spark Structured Streaming test");
+ }
+ }
+
+ public static SparkConf buildSparkConf(String appName, String defaultMaster) {
+ return buildSparkConf(appName, defaultMaster, new HashMap<>());
+ }
+
+ private static SparkConf buildSparkConf(String appName, String defaultMaster, Map<String, String> additionalConfigs) {
+ final SparkConf sparkConf = new SparkConf().setAppName(appName);
+ String master = sparkConf.get("spark.master", defaultMaster);
+ sparkConf.setMaster(master);
+ if (master.startsWith("yarn")) {
+ sparkConf.set("spark.eventLog.overwrite", "true");
+ sparkConf.set("spark.eventLog.enabled", "true");
+ }
+ sparkConf.set("spark.ui.port", "8090");
+ sparkConf.setIfMissing("spark.driver.maxResultSize", "2g");
+ sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
+ sparkConf.set("spark.kryo.registrator", "org.apache.spark.HoodieSparkKryoRegistrar");
+ sparkConf.set("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension");
+ sparkConf.set("spark.hadoop.mapred.output.compress", "true");
+ sparkConf.set("spark.hadoop.mapred.output.compression.codec", "true");
+ sparkConf.set("spark.hadoop.mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec");
+ sparkConf.set("spark.hadoop.mapred.output.compression.type", "BLOCK");
+
+ additionalConfigs.forEach(sparkConf::set);
+ return sparkConf;
+ }
+}
diff --git a/hudi-integ-test/src/main/scala/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkTestWriter.scala b/hudi-integ-test/src/main/scala/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkTestWriter.scala
new file mode 100644
index 00000000000..8eb3b469e93
--- /dev/null
+++ b/hudi-integ-test/src/main/scala/org/apache/hudi/integ/testsuite/streaming/StructuredStreamingSinkTestWriter.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.integ.testsuite.streaming
+
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.config.HoodieWriteConfig.FAIL_ON_TIMELINE_ARCHIVING_ENABLE
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.streaming.StreamingQueryListener.{QueryProgressEvent, QueryStartedEvent, QueryTerminatedEvent}
+import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryListener, Trigger}
+import org.apache.log4j.LogManager
+
+object StructuredStreamingSinkTestWriter {
+
+ private val log = LogManager.getLogger(getClass)
+ var validationComplete: Boolean = false;
+
+ def waitUntilCondition(): Unit = {
+ waitUntilCondition(1000 * 60 * 5, 500)
+ }
+
+ def waitUntilCondition(maxWaitTimeMs: Long, intervalTimeMs: Long): Unit = {
+ var waitSoFar: Long = 0;
+ while (waitSoFar < maxWaitTimeMs && !validationComplete) {
+ log.info("Waiting for " + intervalTimeMs + ". Total wait time " + waitSoFar)
+ Thread.sleep(intervalTimeMs)
+ waitSoFar += intervalTimeMs
+ }
+ }
+
+ def triggerStreaming(spark: SparkSession, tableType: String, inputPath: String, hudiPath: String, hudiCheckpointPath: String,
+ tableName: String, partitionPathField: String, recordKeyField: String,
+ preCombineField: String): Unit = {
+
+ def validate(): Unit = {
+ log.info("Validation starting")
+ val inputDf = spark.read.format("parquet").load(inputPath)
+ val hudiDf = spark.read.format("hudi").load(hudiPath)
+ inputDf.registerTempTable("inputTbl")
+ hudiDf.registerTempTable("hudiTbl")
+ assert(spark.sql("select count(distinct " + partitionPathField + ", " + recordKeyField + ") from inputTbl").count ==
+ spark.sql("select count(distinct " + partitionPathField + ", " + recordKeyField + ") from hudiTbl").count)
+ validationComplete = true
+ log.info("Validation complete")
+ }
+
+ def shutdownListener(spark: SparkSession) = new StreamingQueryListener() {
+ override def onQueryStarted(queryStarted: QueryStartedEvent): Unit = {
+ log.info("Query started: " + queryStarted.id)
+ }
+
+ override def onQueryTerminated(queryTerminated: QueryTerminatedEvent): Unit = {
+ log.info("Query terminated! " + queryTerminated.id + ". Validating input and hudi")
+ validate()
+ log.info("Data Validation complete")
+ }
+
+ override def onQueryProgress(queryProgressEvent: QueryProgressEvent): Unit = {
+ if (queryProgressEvent.progress.numInputRows == 0) {
+ log.info("Stopping spark structured streaming as we have reached the end")
+ spark.streams.active.foreach(_.stop())
+ }
+ }
+ }
+
+ spark.streams.addListener(shutdownListener(spark))
+ log.info("Starting to consume from source and writing to hudi ")
+
+ val inputDfSchema = spark.read.format("parquet").load(inputPath).schema
+ val parquetdf = spark.readStream.option("spark.sql.streaming.schemaInference", "true").option("maxFilesPerTrigger", "1")
+ .schema(inputDfSchema).parquet(inputPath)
+
+ val writer = parquetdf.writeStream.format("org.apache.hudi").
+ option(TABLE_TYPE.key, tableType).
+ option(PRECOMBINE_FIELD.key, preCombineField).
+ option(RECORDKEY_FIELD.key, recordKeyField).
+ option(PARTITIONPATH_FIELD.key, partitionPathField).
+ option(FAIL_ON_TIMELINE_ARCHIVING_ENABLE.key, false).
+ option(STREAMING_IGNORE_FAILED_BATCH.key, false).
+ option(STREAMING_RETRY_CNT.key, 0).
+ option("hoodie.table.name", tableName).
+ option("hoodie.compact.inline.max.delta.commits", "2").
+ option("checkpointLocation", hudiCheckpointPath).
+ outputMode(OutputMode.Append());
+
+ writer.trigger(Trigger.ProcessingTime(30000)).start(hudiPath);
+ }
+}
(hudi) 03/14: [HUDI-7197] Adding mis fixes related with table services testing (#10280)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 4680cb453334e545b3de553f9d98cd3bce829173
Author: harshal <ha...@gmail.com>
AuthorDate: Thu Mar 7 13:52:40 2024 +0530
[HUDI-7197] Adding mis fixes related with table services testing (#10280)
---
.../apache/hudi/table/action/clean/CleanActionExecutor.java | 2 +-
.../hudi/table/action/commit/BaseCommitActionExecutor.java | 1 +
.../main/java/org/apache/hudi/utilities/UtilHelpers.java | 1 +
.../java/org/apache/hudi/utilities/TestHoodieIndexer.java | 13 +++++++++++--
4 files changed, 14 insertions(+), 3 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
index 0b5b3dfa42f..f84dac5fe6f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
@@ -234,7 +234,7 @@ public class CleanActionExecutor<T, I, K, O> extends BaseActionExecutor<T, I, K,
throw new HoodieIOException("Failed to clean up after commit", e);
} finally {
if (!skipLocking) {
- this.txnManager.endTransaction(Option.of(inflightInstant));
+ this.txnManager.endTransaction(Option.ofNullable(inflightInstant));
}
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
index 4f4cc7d9bc7..8def1bf3e8a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
@@ -259,6 +259,7 @@ public abstract class BaseCommitActionExecutor<T, I, K, O, R>
writeMetadata.setPartitionToReplaceFileIds(getPartitionToReplacedFileIds(clusteringPlan, writeMetadata));
commitOnAutoCommit(writeMetadata);
if (!writeMetadata.getCommitMetadata().isPresent()) {
+ LOG.info("Found empty commit metadata for clustering with instant time " + instantTime);
HoodieCommitMetadata commitMetadata = CommitUtils.buildMetadata(writeMetadata.getWriteStats().get(), writeMetadata.getPartitionToReplaceFileIds(),
extraMetadata, operationType, getSchemaToStoreInCommit(), getCommitActionType());
writeMetadata.setCommitMetadata(Option.of(commitMetadata));
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
index d0781849755..35904fb2055 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
@@ -620,6 +620,7 @@ public class UtilHelpers {
} while (ret != 0 && maxRetryCount-- > 0);
} catch (Throwable t) {
LOG.error(errorMessage, t);
+ throw new RuntimeException("Failed in retry", t);
}
return ret;
}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java
index e853d0ca366..9614dd28c1e 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java
@@ -39,6 +39,8 @@ import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieMetadataException;
import org.apache.hudi.metadata.HoodieBackedTableMetadata;
import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import org.apache.hudi.metadata.MetadataPartitionType;
@@ -77,6 +79,7 @@ import static org.apache.hudi.utilities.UtilHelpers.SCHEDULE;
import static org.apache.hudi.utilities.UtilHelpers.SCHEDULE_AND_EXECUTE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implements SparkProvider {
@@ -289,7 +292,10 @@ public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implemen
// start the indexer and validate files index is completely built out
HoodieIndexer indexer = new HoodieIndexer(jsc(), config);
// The catchup won't finish due to inflight delta commit, and this is expected
- assertEquals(-1, indexer.start(0));
+ Throwable cause = assertThrows(RuntimeException.class, () -> indexer.start(0))
+ .getCause();
+ assertTrue(cause instanceof HoodieMetadataException);
+ assertTrue(cause.getMessage().contains("Failed to index partition"));
// Now, make sure that the inflight delta commit happened before the async indexer
// is intact
@@ -365,7 +371,10 @@ public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implemen
config.propsFilePath = propsPath;
// start the indexer and validate index building fails
HoodieIndexer indexer = new HoodieIndexer(jsc(), config);
- assertEquals(-1, indexer.start(0));
+ Throwable cause = assertThrows(RuntimeException.class, () -> indexer.start(0))
+ .getCause();
+ assertTrue(cause instanceof HoodieException);
+ assertTrue(cause.getMessage().contains("Metadata table is not yet initialized"));
// validate table config
metaClient = reload(metaClient);
(hudi) 06/14: [MINOR] Separate HoodieSparkWriterTestBase to reduce duplication (#10832)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 06584c6f76815fdc429d51bab66e44ae16e5fe67
Author: Geser Dugarov <ge...@gmail.com>
AuthorDate: Fri Mar 8 07:52:52 2024 +0700
[MINOR] Separate HoodieSparkWriterTestBase to reduce duplication (#10832)
---
.../apache/hudi/HoodieSparkWriterTestBase.scala | 136 ++++++++++++++++++
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 152 +++------------------
.../apache/hudi/TestHoodieSparkSqlWriterUtc.scala | 2 +-
.../hudi/TestTableSchemaResolverWithSparkSQL.scala | 102 +-------------
4 files changed, 162 insertions(+), 230 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/HoodieSparkWriterTestBase.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/HoodieSparkWriterTestBase.scala
new file mode 100644
index 00000000000..c0c1c2c12bd
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/HoodieSparkWriterTestBase.scala
@@ -0,0 +1,136 @@
+/*
+ * 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.commons.io.FileUtils
+import org.apache.hudi.common.model.{HoodieRecord, HoodieTableType}
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.testutils.HoodieClientTestUtils
+import org.apache.spark.SparkContext
+import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
+import org.apache.spark.sql.{Dataset, Row, SQLContext, SparkSession}
+import org.junit.jupiter.api.{AfterEach, BeforeEach}
+
+import scala.collection.JavaConverters
+
+class HoodieSparkWriterTestBase {
+ var spark: SparkSession = _
+ var sqlContext: SQLContext = _
+ var sc: SparkContext = _
+ var tempPath: java.nio.file.Path = _
+ var tempBootStrapPath: java.nio.file.Path = _
+ var hoodieFooTableName = "hoodie_foo_tbl"
+ var tempBasePath: String = _
+ var commonTableModifier: Map[String, String] = Map()
+
+ case class StringLongTest(uuid: String, ts: Long)
+
+ /**
+ * Setup method running before each test.
+ */
+ @BeforeEach
+ def setUp(): Unit = {
+ initSparkContext()
+ tempPath = java.nio.file.Files.createTempDirectory("hoodie_test_path")
+ tempBootStrapPath = java.nio.file.Files.createTempDirectory("hoodie_test_bootstrap")
+ tempBasePath = tempPath.toAbsolutePath.toString
+ commonTableModifier = getCommonParams(tempPath, hoodieFooTableName, HoodieTableType.COPY_ON_WRITE.name())
+ }
+
+ /**
+ * Tear down method running after each test.
+ */
+ @AfterEach
+ def tearDown(): Unit = {
+ cleanupSparkContexts()
+ FileUtils.deleteDirectory(tempPath.toFile)
+ FileUtils.deleteDirectory(tempBootStrapPath.toFile)
+ }
+
+ /**
+ * Utility method for initializing the spark context.
+ */
+ def initSparkContext(): Unit = {
+ val sparkConf = HoodieClientTestUtils.getSparkConfForTest(getClass.getSimpleName)
+
+ spark = SparkSession.builder()
+ .withExtensions(new HoodieSparkSessionExtension)
+ .config(sparkConf)
+ .getOrCreate()
+
+ sc = spark.sparkContext
+ sc.setLogLevel("ERROR")
+ sqlContext = spark.sqlContext
+ }
+
+ /**
+ * Utility method for cleaning up spark resources.
+ */
+ def cleanupSparkContexts(): Unit = {
+ if (sqlContext != null) {
+ sqlContext.clearCache();
+ sqlContext = null;
+ }
+ if (sc != null) {
+ sc.stop()
+ sc = null
+ }
+ if (spark != null) {
+ spark.close()
+ }
+ }
+
+ /**
+ * Utility method for creating common params for writer.
+ *
+ * @param path Path for hoodie table
+ * @param hoodieFooTableName Name of hoodie table
+ * @param tableType Type of table
+ * @return Map of common params
+ */
+ def getCommonParams(path: java.nio.file.Path, hoodieFooTableName: String, tableType: String): Map[String, String] = {
+ Map("path" -> path.toAbsolutePath.toString,
+ HoodieWriteConfig.TBL_NAME.key -> hoodieFooTableName,
+ "hoodie.insert.shuffle.parallelism" -> "1",
+ "hoodie.upsert.shuffle.parallelism" -> "1",
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType,
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+ DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key -> "org.apache.hudi.keygen.SimpleKeyGenerator")
+ }
+
+ /**
+ * Utility method for dropping all hoodie meta related columns.
+ */
+ def dropMetaFields(df: Dataset[Row]): Dataset[Row] = {
+ df.drop(HoodieRecord.HOODIE_META_COLUMNS.get(0)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(1))
+ .drop(HoodieRecord.HOODIE_META_COLUMNS.get(2)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(3))
+ .drop(HoodieRecord.HOODIE_META_COLUMNS.get(4))
+ }
+
+ /**
+ * Utility method for converting list of Row to list of Seq.
+ *
+ * @param inputList list of Row
+ * @return list of Seq
+ */
+ def convertRowListToSeq(inputList: java.util.List[Row]): Seq[Row] =
+ JavaConverters.asScalaIteratorConverter(inputList.iterator).asScala.toSeq
+
+}
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 d7a1f9331ae..0767d055915 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
@@ -19,10 +19,8 @@ package org.apache.hudi
import org.apache.avro.Schema
import org.apache.commons.io.FileUtils
-import org.apache.hudi.DataSourceWriteOptions._
-import org.apache.hudi.HoodieSparkUtils.gteqSpark3_0
import org.apache.hudi.client.SparkRDDWriteClient
-import org.apache.hudi.common.model._
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieRecord, HoodieRecordPayload, HoodieTableType, WriteOperationType}
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient, TableSchemaResolver}
import org.apache.hudi.common.testutils.HoodieTestDataGenerator
import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieIndexConfig, HoodieWriteConfig}
@@ -30,19 +28,15 @@ import org.apache.hudi.exception.{HoodieException, SchemaCompatibilityException}
import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode
import org.apache.hudi.functional.TestBootstrap
import org.apache.hudi.keygen.{ComplexKeyGenerator, NonpartitionedKeyGenerator, SimpleKeyGenerator}
-import org.apache.hudi.testutils.DataSourceTestUtils
-import org.apache.hudi.testutils.HoodieClientTestUtils.getSparkConfForTest
-import org.apache.spark.SparkContext
+import org.apache.hudi.testutils.{DataSourceTestUtils, HoodieClientTestUtils}
import org.apache.spark.api.java.JavaSparkContext
-import org.apache.spark.sql._
+import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
import org.apache.spark.sql.functions.{expr, lit}
-import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
import org.apache.spark.sql.hudi.command.SqlKeyGenerator
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotNull, assertNull, assertTrue, fail}
-import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
+import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.Arguments.arguments
-import org.junit.jupiter.params.provider._
+import org.junit.jupiter.params.provider.{Arguments, CsvSource, EnumSource, MethodSource, ValueSource}
import org.mockito.ArgumentMatchers.any
import org.mockito.Mockito.{spy, times, verify}
import org.scalatest.Assertions.assertThrows
@@ -52,7 +46,6 @@ import java.io.IOException
import java.time.Instant
import java.util.{Collections, Date, UUID}
import scala.collection.JavaConversions._
-import scala.collection.JavaConverters
/**
* Test suite for SparkSqlWriter class.
@@ -60,113 +53,10 @@ import scala.collection.JavaConverters
* Otherwise UTC tests will generate infinite loops, if there is any initiated test with time zone that is greater then UTC+0.
* The reason is in a saved value in the heap of static {@link org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator.lastInstantTime}.
*/
-class TestHoodieSparkSqlWriter {
- var spark: SparkSession = _
- var sqlContext: SQLContext = _
- var sc: SparkContext = _
- var tempPath: java.nio.file.Path = _
- var tempBootStrapPath: java.nio.file.Path = _
- var hoodieFooTableName = "hoodie_foo_tbl"
- var tempBasePath: String = _
- var commonTableModifier: Map[String, String] = Map()
- case class StringLongTest(uuid: String, ts: Long)
+class TestHoodieSparkSqlWriter extends HoodieSparkWriterTestBase {
/**
- * Setup method running before each test.
- */
- @BeforeEach
- def setUp(): Unit = {
- initSparkContext()
- tempPath = java.nio.file.Files.createTempDirectory("hoodie_test_path")
- tempBootStrapPath = java.nio.file.Files.createTempDirectory("hoodie_test_bootstrap")
- tempBasePath = tempPath.toAbsolutePath.toString
- commonTableModifier = getCommonParams(tempPath, hoodieFooTableName, HoodieTableType.COPY_ON_WRITE.name())
- }
-
- /**
- * Tear down method running after each test.
- */
- @AfterEach
- def tearDown(): Unit = {
- cleanupSparkContexts()
- FileUtils.deleteDirectory(tempPath.toFile)
- FileUtils.deleteDirectory(tempBootStrapPath.toFile)
- }
-
- /**
- * Utility method for initializing the spark context.
- *
- * TODO rebase this onto existing base class to avoid duplication
- */
- def initSparkContext(): Unit = {
- val sparkConf = getSparkConfForTest(getClass.getSimpleName)
-
- spark = SparkSession.builder()
- .withExtensions(new HoodieSparkSessionExtension)
- .config(sparkConf)
- .getOrCreate()
-
- sc = spark.sparkContext
- sc.setLogLevel("ERROR")
- sqlContext = spark.sqlContext
- }
-
- /**
- * Utility method for cleaning up spark resources.
- */
- def cleanupSparkContexts(): Unit = {
- if (sqlContext != null) {
- sqlContext.clearCache();
- sqlContext = null;
- }
- if (sc != null) {
- sc.stop()
- sc = null
- }
- if (spark != null) {
- spark.close()
- }
- }
-
- /**
- * Utility method for dropping all hoodie meta related columns.
- */
- def dropMetaFields(df: Dataset[Row]): Dataset[Row] = {
- df.drop(HoodieRecord.HOODIE_META_COLUMNS.get(0)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(1))
- .drop(HoodieRecord.HOODIE_META_COLUMNS.get(2)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(3))
- .drop(HoodieRecord.HOODIE_META_COLUMNS.get(4))
- }
-
- /**
- * Utility method for creating common params for writer.
- *
- * @param path Path for hoodie table
- * @param hoodieFooTableName Name of hoodie table
- * @param tableType Type of table
- * @return Map of common params
- */
- def getCommonParams(path: java.nio.file.Path, hoodieFooTableName: String, tableType: String): Map[String, String] = {
- Map("path" -> path.toAbsolutePath.toString,
- HoodieWriteConfig.TBL_NAME.key -> hoodieFooTableName,
- "hoodie.insert.shuffle.parallelism" -> "1",
- "hoodie.upsert.shuffle.parallelism" -> "1",
- DataSourceWriteOptions.TABLE_TYPE.key -> tableType,
- DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
- DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
- DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key -> "org.apache.hudi.keygen.SimpleKeyGenerator")
- }
-
- /**
- * Utility method for converting list of Row to list of Seq.
- *
- * @param inputList list of Row
- * @return list of Seq
- */
- def convertRowListToSeq(inputList: java.util.List[Row]): Seq[Row] =
- JavaConverters.asScalaIteratorConverter(inputList.iterator).asScala.toSeq
-
- /**
- * Utility method for performing bulk insert tests.
+ * Local utility method for performing bulk insert tests.
*
* @param sortMode Bulk insert sort mode
* @param populateMetaFields Flag for populating meta fields
@@ -226,12 +116,13 @@ class TestHoodieSparkSqlWriter {
val originals = HoodieWriterUtils.parametersWithWriteDefaults(Map.empty)
val rhsKey = "hoodie.right.hand.side.key"
val rhsVal = "hoodie.right.hand.side.val"
- val modifier = Map(OPERATION.key -> INSERT_OPERATION_OPT_VAL, TABLE_TYPE.key -> MOR_TABLE_TYPE_OPT_VAL, rhsKey -> rhsVal)
+ val modifier = Map(DataSourceWriteOptions.OPERATION.key -> DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
+ DataSourceWriteOptions.TABLE_TYPE.key -> DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, rhsKey -> rhsVal)
val modified = HoodieWriterUtils.parametersWithWriteDefaults(modifier)
val matcher = (k: String, v: String) => modified(k) should be(v)
originals foreach {
- case ("hoodie.datasource.write.operation", _) => matcher("hoodie.datasource.write.operation", INSERT_OPERATION_OPT_VAL)
- case ("hoodie.datasource.write.table.type", _) => matcher("hoodie.datasource.write.table.type", MOR_TABLE_TYPE_OPT_VAL)
+ case ("hoodie.datasource.write.operation", _) => matcher("hoodie.datasource.write.operation", DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ case ("hoodie.datasource.write.table.type", _) => matcher("hoodie.datasource.write.table.type", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
case (`rhsKey`, _) => matcher(rhsKey, rhsVal)
case (k, v) => matcher(k, v)
}
@@ -245,7 +136,7 @@ class TestHoodieSparkSqlWriter {
spark.stop()
val session = SparkSession.builder()
// Here we intentionally remove the "spark.serializer" config to test failure
- .config(getSparkConfForTest("hoodie_test").remove("spark.serializer"))
+ .config(HoodieClientTestUtils.getSparkConfForTest("hoodie_test").remove("spark.serializer"))
.getOrCreate()
try {
val sqlContext = session.sqlContext
@@ -290,7 +181,7 @@ class TestHoodieSparkSqlWriter {
assert(tableAlreadyExistException.getMessage.contains(s"${HoodieWriteConfig.TBL_NAME.key}:\thoodie_bar_tbl\thoodie_foo_tbl"))
//on same path try append with delete operation and different("hoodie_bar_tbl") table name which should throw an exception
- val deleteTableModifier = barTableModifier ++ Map(OPERATION.key -> "delete")
+ val deleteTableModifier = barTableModifier ++ Map(DataSourceWriteOptions.OPERATION.key -> "delete")
val deleteCmdException = intercept[HoodieException](HoodieSparkSqlWriter.write(sqlContext, SaveMode.Append, deleteTableModifier, dataFrame2))
assert(tableAlreadyExistException.getMessage.contains("Config conflict"))
assert(tableAlreadyExistException.getMessage.contains(s"${HoodieWriteConfig.TBL_NAME.key}:\thoodie_bar_tbl\thoodie_foo_tbl"))
@@ -454,7 +345,7 @@ def testBulkInsertForDropPartitionColumn(): Unit = {
val fooTableModifier = commonTableModifier.updated("hoodie.bulkinsert.shuffle.parallelism", "4")
.updated(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL)
.updated(DataSourceWriteOptions.ENABLE_ROW_WRITER.key, "true")
- .updated(INSERT_DROP_DUPS.key, "true")
+ .updated(DataSourceWriteOptions.INSERT_DROP_DUPS.key, "true")
// generate the inserts
val schema = DataSourceTestUtils.getStructTypeExampleSchema
@@ -687,10 +578,11 @@ def testBulkInsertForDropPartitionColumn(): Unit = {
.setBaseFileFormat(fooTableParams.getOrElse(HoodieWriteConfig.BASE_FILE_FORMAT.key,
HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().name))
.setArchiveLogFolder(HoodieTableConfig.ARCHIVELOG_FOLDER.defaultValue())
- .setPayloadClassName(PAYLOAD_CLASS_NAME.key)
- .setPreCombineField(fooTableParams.getOrElse(PRECOMBINE_FIELD.key, PRECOMBINE_FIELD.defaultValue()))
+ .setPayloadClassName(DataSourceWriteOptions.PAYLOAD_CLASS_NAME.key)
+ .setPreCombineField(fooTableParams.getOrElse(DataSourceWriteOptions.PRECOMBINE_FIELD.key, DataSourceWriteOptions.PRECOMBINE_FIELD.defaultValue()))
.setPartitionFields(fooTableParams(DataSourceWriteOptions.PARTITIONPATH_FIELD.key))
- .setKeyGeneratorClassProp(fooTableParams.getOrElse(KEYGENERATOR_CLASS_NAME.key, KEYGENERATOR_CLASS_NAME.defaultValue()))
+ .setKeyGeneratorClassProp(fooTableParams.getOrElse(DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key,
+ DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.defaultValue()))
if(addBootstrapPath) {
tableMetaClientBuilder
.setBootstrapBasePath(fooTableParams(HoodieBootstrapConfig.BASE_PATH.key))
@@ -1364,19 +1256,19 @@ object TestHoodieSparkSqlWriter {
// NOTE: Hudi doesn't support Orc in Spark < 3.0
// Please check HUDI-4496 for more details
- val targetScenarios = if (gteqSpark3_0) {
+ val targetScenarios = if (HoodieSparkUtils.gteqSpark3_0) {
parquetScenarios ++ orcScenarios
} else {
parquetScenarios
}
- java.util.Arrays.stream(targetScenarios.map(as => arguments(as.map(_.asInstanceOf[AnyRef]):_*)))
+ java.util.Arrays.stream(targetScenarios.map(as => Arguments.arguments(as.map(_.asInstanceOf[AnyRef]):_*)))
}
def deletePartitionsWildcardTestParams(): java.util.stream.Stream[Arguments] = {
java.util.stream.Stream.of(
- arguments("*5/03/1*", Seq("2016/03/15")),
- arguments("2016/03/*", Seq("2015/03/16", "2015/03/17")))
+ Arguments.arguments("*5/03/1*", Seq("2016/03/15")),
+ Arguments.arguments("2016/03/*", Seq("2015/03/16", "2015/03/17")))
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala
index df8614f5e2a..ca4d23f719d 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala
@@ -36,7 +36,7 @@ import java.util.TimeZone
* value of static {@link HoodieInstantTimeGenerator.lastInstantTime} in the heap,
* which will be greater than instant time for {@link HoodieTimelineTimeZone.UTC}.
*/
-class TestHoodieSparkSqlWriterUtc extends TestHoodieSparkSqlWriter {
+class TestHoodieSparkSqlWriterUtc extends HoodieSparkWriterTestBase {
/*
* Test case for instant is generated with commit timezone when TIMELINE_TIMEZONE set to UTC
* related to HUDI-5978
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestTableSchemaResolverWithSparkSQL.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestTableSchemaResolverWithSparkSQL.scala
index d9d5b59c8d7..70886d96444 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestTableSchemaResolverWithSparkSQL.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestTableSchemaResolverWithSparkSQL.scala
@@ -18,120 +18,24 @@
package org.apache.hudi
import org.apache.avro.Schema
-import org.apache.commons.io.FileUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hudi.avro.HoodieAvroUtils
import org.apache.hudi.avro.model.HoodieMetadataRecord
-import org.apache.hudi.common.model._
import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.testutils.DataSourceTestUtils
-import org.apache.hudi.testutils.HoodieClientTestUtils.getSparkConfForTest
-import org.apache.spark.SparkContext
-import org.apache.spark.sql._
-import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
+import org.apache.spark.sql.SaveMode
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
-import org.junit.jupiter.api.{AfterEach, BeforeEach, Tag, Test}
+import org.junit.jupiter.api.{Tag, Test}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.CsvSource
-import scala.collection.JavaConverters
-
/**
* Test suite for TableSchemaResolver with SparkSqlWriter.
*/
@Tag("functional")
-class TestTableSchemaResolverWithSparkSQL {
- var spark: SparkSession = _
- var sqlContext: SQLContext = _
- var sc: SparkContext = _
- var tempPath: java.nio.file.Path = _
- var tempBootStrapPath: java.nio.file.Path = _
- var hoodieFooTableName = "hoodie_foo_tbl"
- var tempBasePath: String = _
- var commonTableModifier: Map[String, String] = Map()
-
- case class StringLongTest(uuid: String, ts: Long)
-
- /**
- * Setup method running before each test.
- */
- @BeforeEach
- def setUp(): Unit = {
- initSparkContext()
- tempPath = java.nio.file.Files.createTempDirectory("hoodie_test_path")
- tempBootStrapPath = java.nio.file.Files.createTempDirectory("hoodie_test_bootstrap")
- tempBasePath = tempPath.toAbsolutePath.toString
- commonTableModifier = getCommonParams(tempPath, hoodieFooTableName, HoodieTableType.COPY_ON_WRITE.name())
- }
-
- /**
- * Tear down method running after each test.
- */
- @AfterEach
- def tearDown(): Unit = {
- cleanupSparkContexts()
- FileUtils.deleteDirectory(tempPath.toFile)
- FileUtils.deleteDirectory(tempBootStrapPath.toFile)
- }
-
- /**
- * Utility method for initializing the spark context.
- */
- def initSparkContext(): Unit = {
- spark = SparkSession.builder()
- .config(getSparkConfForTest(hoodieFooTableName))
- .getOrCreate()
- sc = spark.sparkContext
- sc.setLogLevel("ERROR")
- sqlContext = spark.sqlContext
- }
-
- /**
- * Utility method for cleaning up spark resources.
- */
- def cleanupSparkContexts(): Unit = {
- if (sqlContext != null) {
- sqlContext.clearCache();
- sqlContext = null;
- }
- if (sc != null) {
- sc.stop()
- sc = null
- }
- if (spark != null) {
- spark.close()
- }
- }
-
- /**
- * Utility method for creating common params for writer.
- *
- * @param path Path for hoodie table
- * @param hoodieFooTableName Name of hoodie table
- * @param tableType Type of table
- * @return Map of common params
- */
- def getCommonParams(path: java.nio.file.Path, hoodieFooTableName: String, tableType: String): Map[String, String] = {
- Map("path" -> path.toAbsolutePath.toString,
- HoodieWriteConfig.TBL_NAME.key -> hoodieFooTableName,
- "hoodie.insert.shuffle.parallelism" -> "1",
- "hoodie.upsert.shuffle.parallelism" -> "1",
- DataSourceWriteOptions.TABLE_TYPE.key -> tableType,
- DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
- DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
- DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key -> "org.apache.hudi.keygen.SimpleKeyGenerator")
- }
-
- /**
- * Utility method for converting list of Row to list of Seq.
- *
- * @param inputList list of Row
- * @return list of Seq
- */
- def convertRowListToSeq(inputList: java.util.List[Row]): Seq[Row] =
- JavaConverters.asScalaIteratorConverter(inputList.iterator).asScala.toSeq
+class TestTableSchemaResolverWithSparkSQL extends HoodieSparkWriterTestBase {
@Test
def testTableSchemaResolverInMetadataTable(): Unit = {
(hudi) 13/14: [HUDI-7163] Fix not parsable text DateTimeParseException when compact (#10220)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 80990d4232577a6e7dbf79fde2e312fda6d9ddcc
Author: wuzhenhua <10...@users.noreply.github.com>
AuthorDate: Sun Mar 10 04:49:54 2024 +0800
[HUDI-7163] Fix not parsable text DateTimeParseException when compact (#10220)
---
.../compact/ScheduleCompactionActionExecutor.java | 9 ++---
.../java/org/apache/hudi/util/StreamerUtil.java | 39 ++++++++++------------
2 files changed, 19 insertions(+), 29 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
index f529285e29d..e7d1138fd77 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
@@ -236,12 +236,7 @@ public class ScheduleCompactionActionExecutor<T, I, K, O> extends BaseActionExec
}
private Long parsedToSeconds(String time) {
- long timestamp;
- try {
- timestamp = HoodieActiveTimeline.parseDateFromInstantTime(time).getTime() / 1000;
- } catch (ParseException e) {
- throw new HoodieCompactionException(e.getMessage(), e);
- }
- return timestamp;
+ return HoodieActiveTimeline.parseDateFromInstantTimeSafely(time).orElseThrow(() -> new HoodieCompactionException("Failed to parse timestamp " + time))
+ .getTime() / 1000;
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
index 40519ae4ed7..176ba61b2b1 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
@@ -65,7 +65,6 @@ import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.StringReader;
-import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
@@ -328,34 +327,30 @@ public class StreamerUtil {
* Returns the median instant time between the given two instant time.
*/
public static Option<String> medianInstantTime(String highVal, String lowVal) {
- try {
- long high = HoodieActiveTimeline.parseDateFromInstantTime(highVal).getTime();
- long low = HoodieActiveTimeline.parseDateFromInstantTime(lowVal).getTime();
- ValidationUtils.checkArgument(high > low,
- "Instant [" + highVal + "] should have newer timestamp than instant [" + lowVal + "]");
- long median = low + (high - low) / 2;
- final String instantTime = HoodieActiveTimeline.formatDate(new Date(median));
- if (HoodieTimeline.compareTimestamps(lowVal, HoodieTimeline.GREATER_THAN_OR_EQUALS, instantTime)
- || HoodieTimeline.compareTimestamps(highVal, HoodieTimeline.LESSER_THAN_OR_EQUALS, instantTime)) {
- return Option.empty();
- }
- return Option.of(instantTime);
- } catch (ParseException e) {
- throw new HoodieException("Get median instant time with interval [" + lowVal + ", " + highVal + "] error", e);
+ long high = HoodieActiveTimeline.parseDateFromInstantTimeSafely(highVal)
+ .orElseThrow(() -> new HoodieException("Get instant time diff with interval [" + highVal + "] error")).getTime();
+ long low = HoodieActiveTimeline.parseDateFromInstantTimeSafely(lowVal)
+ .orElseThrow(() -> new HoodieException("Get instant time diff with interval [" + lowVal + "] error")).getTime();
+ ValidationUtils.checkArgument(high > low,
+ "Instant [" + highVal + "] should have newer timestamp than instant [" + lowVal + "]");
+ long median = low + (high - low) / 2;
+ final String instantTime = HoodieActiveTimeline.formatDate(new Date(median));
+ if (HoodieTimeline.compareTimestamps(lowVal, HoodieTimeline.GREATER_THAN_OR_EQUALS, instantTime)
+ || HoodieTimeline.compareTimestamps(highVal, HoodieTimeline.LESSER_THAN_OR_EQUALS, instantTime)) {
+ return Option.empty();
}
+ return Option.of(instantTime);
}
/**
* Returns the time interval in seconds between the given instant time.
*/
public static long instantTimeDiffSeconds(String newInstantTime, String oldInstantTime) {
- try {
- long newTimestamp = HoodieActiveTimeline.parseDateFromInstantTime(newInstantTime).getTime();
- long oldTimestamp = HoodieActiveTimeline.parseDateFromInstantTime(oldInstantTime).getTime();
- return (newTimestamp - oldTimestamp) / 1000;
- } catch (ParseException e) {
- throw new HoodieException("Get instant time diff with interval [" + oldInstantTime + ", " + newInstantTime + "] error", e);
- }
+ long newTimestamp = HoodieActiveTimeline.parseDateFromInstantTimeSafely(newInstantTime)
+ .orElseThrow(() -> new HoodieException("Get instant time diff with interval [" + oldInstantTime + ", " + newInstantTime + "] error")).getTime();
+ long oldTimestamp = HoodieActiveTimeline.parseDateFromInstantTimeSafely(oldInstantTime)
+ .orElseThrow(() -> new HoodieException("Get instant time diff with interval [" + oldInstantTime + ", " + newInstantTime + "] error")).getTime();
+ return (newTimestamp - oldTimestamp) / 1000;
}
public static Option<Transformer> createTransformer(List<String> classNames) throws IOException {
(hudi) 04/14: [HUDI-5167] Reducing total test run time: reducing tests for virtual keys (#7153)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 9f00f6d6ed7629d2e73d82fcab66d329b5487c43
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Thu Mar 7 06:54:43 2024 -0800
[HUDI-5167] Reducing total test run time: reducing tests for virtual keys (#7153)
---
.../TestHoodieClientOnCopyOnWriteStorage.java | 112 +++++++++------------
1 file changed, 49 insertions(+), 63 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
index 1b7948eb284..eddded4d6c8 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
@@ -206,7 +206,6 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
private static Stream<Arguments> rollbackFailedCommitsParams() {
return Stream.of(
Arguments.of(HoodieFailedWritesCleaningPolicy.LAZY, true),
- Arguments.of(HoodieFailedWritesCleaningPolicy.LAZY, false),
Arguments.of(HoodieFailedWritesCleaningPolicy.NEVER, true),
Arguments.of(HoodieFailedWritesCleaningPolicy.NEVER, false)
);
@@ -242,10 +241,9 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test Auto Commit behavior for HoodieWriteClient insertPrepped API.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testAutoCommitOnInsertPrepped(boolean populateMetaFields) throws Exception {
- testAutoCommit(SparkRDDWriteClient::insertPreppedRecords, true, populateMetaFields);
+ @Test
+ public void testAutoCommitOnInsertPrepped() throws Exception {
+ testAutoCommit(SparkRDDWriteClient::insertPreppedRecords, true, true);
}
/**
@@ -278,11 +276,10 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test Auto Commit behavior for HoodieWriteClient bulk-insert prepped API.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testAutoCommitOnBulkInsertPrepped(boolean populateMetaFields) throws Exception {
+ @Test
+ public void testAutoCommitOnBulkInsertPrepped() throws Exception {
testAutoCommit((writeClient, recordRDD, instantTime) -> writeClient.bulkInsertPreppedRecords(recordRDD, instantTime,
- Option.empty()), true, populateMetaFields);
+ Option.empty()), true, true);
}
/**
@@ -442,10 +439,9 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test De-duplication behavior for HoodieWriteClient upsert API.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testDeduplicationOnUpsert(boolean populateMetaFields) throws Exception {
- testDeduplication(SparkRDDWriteClient::upsert, populateMetaFields);
+ @Test
+ public void testDeduplicationOnUpsert() throws Exception {
+ testDeduplication(SparkRDDWriteClient::upsert, true);
}
/**
@@ -600,11 +596,10 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test UpsertPrepped API.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testUpsertsPrepped(boolean populateMetaFields) throws Exception {
+ @Test
+ public void testUpsertsPrepped() throws Exception {
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder().withRollbackUsingMarkers(true);
- addConfigsForPopulateMetaFields(cfgBuilder, populateMetaFields);
+ addConfigsForPopulateMetaFields(cfgBuilder, true);
testUpsertsInternal(cfgBuilder.build(), SparkRDDWriteClient::upsertPreppedRecords, true);
}
@@ -839,11 +834,10 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test InsertPrepped API for HoodieConcatHandle.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testInsertsPreppedWithHoodieConcatHandle(boolean populateMetaFields) throws Exception {
+ @Test
+ public void testInsertsPreppedWithHoodieConcatHandle() throws Exception {
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder();
- addConfigsForPopulateMetaFields(cfgBuilder, populateMetaFields);
+ addConfigsForPopulateMetaFields(cfgBuilder, true);
testHoodieConcatHandle(cfgBuilder.build(), true);
}
@@ -997,11 +991,10 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Tests deletion of records.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testDeletes(boolean populateMetaFields) throws Exception {
+ @Test
+ public void testDeletes() throws Exception {
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder(HoodieFailedWritesCleaningPolicy.LAZY);
- addConfigsForPopulateMetaFields(cfgBuilder, populateMetaFields);
+ addConfigsForPopulateMetaFields(cfgBuilder, true);
SparkRDDWriteClient client = getHoodieWriteClient(cfgBuilder.build());
/**
* Write 1 (inserts and deletes) Write actual 200 insert records and ignore 100 delete records
@@ -1022,7 +1015,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
writeBatch(client, newCommitTime, initCommitTime, Option.empty(), initCommitTime,
// unused as genFn uses hard-coded number of inserts/updates/deletes
-1, recordGenFunction, SparkRDDWriteClient::upsert, true, 200, 200, 1, false,
- populateMetaFields);
+ true);
/**
* Write 2 (deletes+writes).
@@ -1040,7 +1033,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
};
writeBatch(client, newCommitTime, prevCommitTime, Option.empty(), initCommitTime, 100, recordGenFunction,
SparkRDDWriteClient::upsert, true, 50, 150, 2, false,
- populateMetaFields);
+ true);
}
/**
@@ -1049,11 +1042,10 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
*
* @throws Exception
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testDeletesForInsertsInSameBatch(boolean populateMetaFields) throws Exception {
+ @Test
+ public void testDeletesForInsertsInSameBatch() throws Exception {
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder(HoodieFailedWritesCleaningPolicy.LAZY);
- addConfigsForPopulateMetaFields(cfgBuilder, populateMetaFields);
+ addConfigsForPopulateMetaFields(cfgBuilder, true);
SparkRDDWriteClient client = getHoodieWriteClient(cfgBuilder.build());
/**
* Write 200 inserts and issue deletes to a subset(50) of inserts.
@@ -1074,7 +1066,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
writeBatch(client, newCommitTime, initCommitTime, Option.empty(), initCommitTime,
-1, recordGenFunction, SparkRDDWriteClient::upsert, true, 150, 150, 1, false,
- populateMetaFields);
+ true);
}
private void assertPartitionPathRecordKeys(List<Pair<String, String>> expectedPartitionPathRecKeyPairs, String[] fullPartitionPaths) {
@@ -1903,19 +1895,17 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test scenario of writing fewer file groups than existing number of file groups in partition.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testInsertOverwritePartitionHandlingWithFewerRecords(boolean populateMetaFields) throws Exception {
- verifyInsertOverwritePartitionHandling(3000, 1000, populateMetaFields);
+ @Test
+ public void testInsertOverwritePartitionHandlingWithFewerRecords() throws Exception {
+ verifyInsertOverwritePartitionHandling(3000, 1000, true);
}
/**
* Test scenario of writing similar number file groups in partition.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testInsertOverwritePartitionHandlingWithSimilarNumberOfRecords(boolean populateMetaFields) throws Exception {
- verifyInsertOverwritePartitionHandling(3000, 3000, populateMetaFields);
+ @Test
+ public void testInsertOverwritePartitionHandlingWithSimilarNumberOfRecords() throws Exception {
+ verifyInsertOverwritePartitionHandling(3000, 3000, true);
}
/**
@@ -1968,19 +1958,17 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test scenario of writing similar number file groups in partition.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void verifyDeletePartitionsHandlingWithSimilarNumberOfRecords(boolean populateMetaFields) throws Exception {
- verifyDeletePartitionsHandling(3000, 3000, 3000, populateMetaFields);
+ @Test
+ public void verifyDeletePartitionsHandlingWithSimilarNumberOfRecords() throws Exception {
+ verifyDeletePartitionsHandling(3000, 3000, 3000, true);
}
/**
* Test scenario of writing more file groups for first partition than second and third partition.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void verifyDeletePartitionsHandlingHandlingWithFewerRecordsSecondThirdPartition(boolean populateMetaFields) throws Exception {
- verifyDeletePartitionsHandling(3000, 1000, 1000, populateMetaFields);
+ @Test
+ public void verifyDeletePartitionsHandlingHandlingWithFewerRecordsSecondThirdPartition() throws Exception {
+ verifyDeletePartitionsHandling(3000, 1000, 1000, true);
}
private Set<String> insertPartitionRecordsWithCommit(SparkRDDWriteClient client, int recordsCount, String commitTime1, String partitionPath) throws IOException {
@@ -2222,13 +2210,12 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
/**
* Test to ensure commit metadata points to valid files.
*/
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testCommitWritesRelativePaths(boolean populateMetaFields) throws Exception {
+ @Test
+ public void testCommitWritesRelativePaths() throws Exception {
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder().withAutoCommit(false);
- addConfigsForPopulateMetaFields(cfgBuilder, populateMetaFields);
- try (SparkRDDWriteClient client = getHoodieWriteClient(cfgBuilder.build());) {
+ addConfigsForPopulateMetaFields(cfgBuilder, true);
+ try (SparkRDDWriteClient client = getHoodieWriteClient(cfgBuilder.build())) {
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(basePath).build();
HoodieSparkTable table = HoodieSparkTable.create(cfgBuilder.build(), context, metaClient);
@@ -2403,9 +2390,9 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
}
@ParameterizedTest
- @MethodSource("rollbackAfterConsistencyCheckFailureParams")
- public void testRollbackAfterConsistencyCheckFailureUsingFileList(boolean enableOptimisticConsistencyGuard, boolean populateMetCols) throws Exception {
- testRollbackAfterConsistencyCheckFailureUsingFileList(false, enableOptimisticConsistencyGuard, populateMetCols);
+ @ValueSource(booleans = {true, false})
+ public void testRollbackAfterConsistencyCheckFailureUsingFileList(boolean enableOptimisticConsistencyGuard) throws Exception {
+ testRollbackAfterConsistencyCheckFailureUsingFileList(false, enableOptimisticConsistencyGuard, true);
}
@ParameterizedTest
@@ -2496,9 +2483,9 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
}
}
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testRollbackFailedCommitsToggleCleaningPolicy(boolean populateMetaFields) throws Exception {
+ @Test
+ public void testRollbackFailedCommitsToggleCleaningPolicy() throws Exception {
+ boolean populateMetaFields = true;
HoodieTestUtils.init(hadoopConf, basePath);
HoodieFailedWritesCleaningPolicy cleaningPolicy = EAGER;
SparkRDDWriteClient client = new SparkRDDWriteClient(context, getParallelWritingWriteConfig(cleaningPolicy, populateMetaFields));
@@ -2665,12 +2652,11 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
return Pair.of(markerFilePath.get(), result);
}
- @ParameterizedTest
- @MethodSource("populateMetaFieldsParams")
- public void testMultiOperationsPerCommit(boolean populateMetaFields) throws IOException {
+ @Test
+ public void testMultiOperationsPerCommit() throws IOException {
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder().withAutoCommit(false)
.withAllowMultiWriteOnSameInstant(true);
- addConfigsForPopulateMetaFields(cfgBuilder, populateMetaFields);
+ addConfigsForPopulateMetaFields(cfgBuilder, true);
HoodieWriteConfig cfg = cfgBuilder.build();
SparkRDDWriteClient client = getHoodieWriteClient(cfg);
String firstInstantTime = "0000";
(hudi) 02/14: [HUDI-7356] Passing configs to file reader constructor for flexibility (#10698)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 695577bdc958c4edf7a81b306ea75ab0d3116c03
Author: wombatu-kun <wo...@gmail.com>
AuthorDate: Thu Mar 7 12:31:56 2024 +0700
[HUDI-7356] Passing configs to file reader constructor for flexibility (#10698)
Co-authored-by: Vova Kolmakov <ko...@huawei-partners.com>
---
.../io/storage/HoodieSparkFileReaderFactory.java | 9 +++++++--
.../hudi/io/storage/HoodieAvroFileReaderFactory.java | 13 +++++++++----
.../hudi/io/storage/HoodieFileReaderFactory.java | 20 ++++++++++----------
3 files changed, 26 insertions(+), 16 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
index f981061ecc3..d06b6913905 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
@@ -18,6 +18,7 @@
package org.apache.hudi.io.storage;
+import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieIOException;
@@ -30,7 +31,8 @@ import java.io.IOException;
public class HoodieSparkFileReaderFactory extends HoodieFileReaderFactory {
- protected HoodieFileReader newParquetFileReader(Configuration conf, Path path) {
+ @Override
+ public HoodieFileReader newParquetFileReader(Configuration conf, Path path) {
conf.setIfUnset(SQLConf.PARQUET_BINARY_AS_STRING().key(), SQLConf.PARQUET_BINARY_AS_STRING().defaultValueString());
conf.setIfUnset(SQLConf.PARQUET_INT96_AS_TIMESTAMP().key(), SQLConf.PARQUET_INT96_AS_TIMESTAMP().defaultValueString());
conf.setIfUnset(SQLConf.CASE_SENSITIVE().key(), SQLConf.CASE_SENSITIVE().defaultValueString());
@@ -42,12 +44,15 @@ public class HoodieSparkFileReaderFactory extends HoodieFileReaderFactory {
return new HoodieSparkParquetReader(conf, path);
}
- protected HoodieFileReader newHFileFileReader(Configuration conf,
+ @Override
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
+ Configuration conf,
Path path,
Option<Schema> schemaOption) throws IOException {
throw new HoodieIOException("Not support read HFile");
}
+ @Override
protected HoodieFileReader newOrcFileReader(Configuration conf, Path path) {
throw new HoodieIOException("Not support read orc file");
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileReaderFactory.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileReaderFactory.java
index 0a511d10b03..84aed905a4d 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileReaderFactory.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileReaderFactory.java
@@ -18,6 +18,7 @@
package org.apache.hudi.io.storage;
+import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.util.Option;
import org.apache.avro.Schema;
@@ -29,15 +30,18 @@ import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import java.io.IOException;
public class HoodieAvroFileReaderFactory extends HoodieFileReaderFactory {
+
+ @Override
protected HoodieFileReader newParquetFileReader(Configuration conf, Path path) {
return new HoodieAvroParquetReader(conf, path);
}
- protected HoodieFileReader newHFileFileReader(boolean useNativeHFileReader,
+ @Override
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
Configuration conf,
Path path,
Option<Schema> schemaOption) throws IOException {
- if (useNativeHFileReader) {
+ if (isUseNativeHFileReaderEnabled(hoodieConfig)) {
return new HoodieNativeAvroHFileReader(conf, path, schemaOption);
}
CacheConfig cacheConfig = new CacheConfig(conf);
@@ -47,14 +51,15 @@ public class HoodieAvroFileReaderFactory extends HoodieFileReaderFactory {
return new HoodieHBaseAvroHFileReader(conf, path, cacheConfig);
}
- protected HoodieFileReader newHFileFileReader(boolean useNativeHFileReader,
+ @Override
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
Configuration conf,
Path path,
FileSystem fs,
byte[] content,
Option<Schema> schemaOption)
throws IOException {
- if (useNativeHFileReader) {
+ if (isUseNativeHFileReaderEnabled(hoodieConfig)) {
return new HoodieNativeAvroHFileReader(conf, content, schemaOption);
}
CacheConfig cacheConfig = new CacheConfig(conf);
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java
index f4b4bedc468..ac2736f8829 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieFileReaderFactory.java
@@ -84,11 +84,9 @@ public class HoodieFileReaderFactory {
Option<Schema> schemaOption) throws IOException {
switch (format) {
case PARQUET:
- return this.newParquetFileReader(conf, path);
+ return newParquetFileReader(conf, path);
case HFILE:
- boolean useNativeHFileReader =
- hoodieConfig.getBooleanOrDefault(HoodieReaderConfig.USE_NATIVE_HFILE_READER);
- return newHFileFileReader(useNativeHFileReader, conf, path, schemaOption);
+ return newHFileFileReader(hoodieConfig, conf, path, schemaOption);
case ORC:
return newOrcFileReader(conf, path);
default:
@@ -96,15 +94,13 @@ public class HoodieFileReaderFactory {
}
}
- public HoodieFileReader getContentReader(HoodieConfig config,
+ public HoodieFileReader getContentReader(HoodieConfig hoodieConfig,
Configuration conf, Path path, HoodieFileFormat format,
FileSystem fs, byte[] content,
Option<Schema> schemaOption) throws IOException {
switch (format) {
case HFILE:
- boolean useNativeHFileReader =
- config.getBooleanOrDefault(HoodieReaderConfig.USE_NATIVE_HFILE_READER);
- return newHFileFileReader(useNativeHFileReader, conf, path, fs, content, schemaOption);
+ return newHFileFileReader(hoodieConfig, conf, path, fs, content, schemaOption);
default:
throw new UnsupportedOperationException(format + " format not supported yet.");
}
@@ -114,13 +110,13 @@ public class HoodieFileReaderFactory {
throw new UnsupportedOperationException();
}
- protected HoodieFileReader newHFileFileReader(boolean useNativeHFileReader,
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
Configuration conf, Path path,
Option<Schema> schemaOption) throws IOException {
throw new UnsupportedOperationException();
}
- protected HoodieFileReader newHFileFileReader(boolean useNativeHFileReader,
+ protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
Configuration conf, Path path,
FileSystem fs,
byte[] content, Option<Schema> schemaOption)
@@ -138,4 +134,8 @@ public class HoodieFileReaderFactory {
Object[] partitionValues) {
throw new UnsupportedOperationException();
}
+
+ protected static boolean isUseNativeHFileReaderEnabled(HoodieConfig hoodieConfig) {
+ return hoodieConfig.getBooleanOrDefault(HoodieReaderConfig.USE_NATIVE_HFILE_READER);
+ }
}
(hudi) 01/14: [HUDI-6947] Refactored HoodieSchemaUtils.deduceWriterSchema with many flags (#10810)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 45923f3cacd709ce60d753a58e6f37a9759cbf19
Author: Geser Dugarov <ge...@gmail.com>
AuthorDate: Thu Mar 7 12:23:38 2024 +0700
[HUDI-6947] Refactored HoodieSchemaUtils.deduceWriterSchema with many flags (#10810)
---
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 176 +++++++++++----------
1 file changed, 93 insertions(+), 83 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
index cfc43453e9c..9aeff64f237 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
@@ -76,107 +76,117 @@ object HoodieSchemaUtils {
latestTableSchemaOpt: Option[Schema],
internalSchemaOpt: Option[InternalSchema],
opts: Map[String, String]): Schema = {
- val setNullForMissingColumns = opts.getOrDefault(DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.key(),
- DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.defaultValue).toBoolean
- val shouldReconcileSchema = opts.getOrDefault(DataSourceWriteOptions.RECONCILE_SCHEMA.key(),
- DataSourceWriteOptions.RECONCILE_SCHEMA.defaultValue().toString).toBoolean
- val shouldValidateSchemasCompatibility = opts.getOrDefault(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key,
- HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.defaultValue).toBoolean
-
latestTableSchemaOpt match {
- // In case table schema is empty we're just going to use the source schema as a
- // writer's schema.
+ // If table schema is empty, then we use the source schema as a writer's schema.
case None => AvroInternalSchemaConverter.fixNullOrdering(sourceSchema)
// Otherwise, we need to make sure we reconcile incoming and latest table schemas
case Some(latestTableSchemaWithMetaFields) =>
- // NOTE: Meta-fields will be unconditionally injected by Hudi writing handles, for the sake of
- // deducing proper writer schema we're stripping them to make sure we can perform proper
- // analysis
- //add call to fix null ordering to ensure backwards compatibility
+ // NOTE: Meta-fields will be unconditionally injected by Hudi writing handles, for the sake of deducing proper writer schema
+ // we're stripping them to make sure we can perform proper analysis
+ // add call to fix null ordering to ensure backwards compatibility
val latestTableSchema = AvroInternalSchemaConverter.fixNullOrdering(removeMetadataFields(latestTableSchemaWithMetaFields))
+
// Before validating whether schemas are compatible, we need to "canonicalize" source's schema
// relative to the table's one, by doing a (minor) reconciliation of the nullability constraints:
// for ex, if in incoming schema column A is designated as non-null, but it's designated as nullable
// in the table's one we want to proceed aligning nullability constraints w/ the table's schema
// Also, we promote types to the latest table schema if possible.
- val shouldCanonicalizeSchema = opts.getOrDefault(CANONICALIZE_SCHEMA.key,
- CANONICALIZE_SCHEMA.defaultValue.toString).toBoolean
- val mergeIntoWrites = opts.getOrDefault(SQL_MERGE_INTO_WRITES.key(),
- SQL_MERGE_INTO_WRITES.defaultValue.toString).toBoolean
-
+ val shouldCanonicalizeSchema = opts.getOrDefault(CANONICALIZE_SCHEMA.key, CANONICALIZE_SCHEMA.defaultValue.toString).toBoolean
val canonicalizedSourceSchema = if (shouldCanonicalizeSchema) {
canonicalizeSchema(sourceSchema, latestTableSchema, opts)
} else {
AvroInternalSchemaConverter.fixNullOrdering(sourceSchema)
}
- val allowAutoEvolutionColumnDrop = opts.getOrDefault(HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.key,
- HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.defaultValue).toBoolean
-
+ val shouldReconcileSchema = opts.getOrDefault(DataSourceWriteOptions.RECONCILE_SCHEMA.key(),
+ DataSourceWriteOptions.RECONCILE_SCHEMA.defaultValue().toString).toBoolean
if (shouldReconcileSchema) {
- internalSchemaOpt match {
- case Some(internalSchema) =>
- // Apply schema evolution, by auto-merging write schema and read schema
- val mergedInternalSchema = AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, internalSchema)
- val evolvedSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema, latestTableSchema.getFullName)
- val shouldRemoveMetaDataFromInternalSchema = sourceSchema.getFields().filter(f => f.name().equalsIgnoreCase(HoodieRecord.RECORD_KEY_METADATA_FIELD)).isEmpty
- if (shouldRemoveMetaDataFromInternalSchema) HoodieAvroUtils.removeMetadataFields(evolvedSchema) else evolvedSchema
- case None =>
- // In case schema reconciliation is enabled we will employ (legacy) reconciliation
- // strategy to produce target writer's schema (see definition below)
- val (reconciledSchema, isCompatible) =
- reconcileSchemasLegacy(latestTableSchema, canonicalizedSourceSchema)
-
- // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
- // w/ the table's one and allow schemas to diverge. This is required in cases where
- // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
- // only incoming dataset's projection has to match the table's schema, and not the whole one
- if (!shouldValidateSchemasCompatibility || isCompatible) {
- reconciledSchema
- } else {
- log.error(
- s"""Failed to reconcile incoming batch schema with the table's one.
- |Incoming schema ${sourceSchema.toString(true)}
- |Incoming schema (canonicalized) ${canonicalizedSourceSchema.toString(true)}
- |Table's schema ${latestTableSchema.toString(true)}
- |""".stripMargin)
- throw new SchemaCompatibilityException("Failed to reconcile incoming schema with the table's one")
- }
- }
+ deduceWriterSchemaWithReconcile(sourceSchema, canonicalizedSourceSchema, latestTableSchema, internalSchemaOpt, opts)
+ } else {
+ deduceWriterSchemaWithoutReconcile(sourceSchema, canonicalizedSourceSchema, latestTableSchema, opts)
+ }
+ }
+ }
+
+ /**
+ * Deducing with disabled reconciliation.
+ * We have to validate that the source's schema is compatible w/ the table's latest schema,
+ * such that we're able to read existing table's records using [[sourceSchema]].
+ */
+ private def deduceWriterSchemaWithoutReconcile(sourceSchema: Schema,
+ canonicalizedSourceSchema: Schema,
+ latestTableSchema: Schema,
+ opts: Map[String, String]): Schema = {
+ // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
+ // w/ the table's one and allow schemas to diverge. This is required in cases where
+ // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
+ // only incoming dataset's projection has to match the table's schema, and not the whole one
+ val mergeIntoWrites = opts.getOrDefault(SQL_MERGE_INTO_WRITES.key(), SQL_MERGE_INTO_WRITES.defaultValue.toString).toBoolean
+ val shouldValidateSchemasCompatibility = opts.getOrDefault(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key,
+ HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.defaultValue).toBoolean
+ val allowAutoEvolutionColumnDrop = opts.getOrDefault(HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.key,
+ HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.defaultValue).toBoolean
+ val setNullForMissingColumns = opts.getOrDefault(DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.key(),
+ DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.defaultValue).toBoolean
+
+ if (!mergeIntoWrites && !shouldValidateSchemasCompatibility && !allowAutoEvolutionColumnDrop) {
+ // Default behaviour
+ val reconciledSchema = if (setNullForMissingColumns) {
+ AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, latestTableSchema)
+ } else {
+ canonicalizedSourceSchema
+ }
+ checkValidEvolution(reconciledSchema, latestTableSchema)
+ reconciledSchema
+ } else {
+ // If it's merge into writes, we don't check for projection nor schema compatibility. Writers down the line will take care of it.
+ // Or it's not merge into writes, and we don't validate schema, but we allow to drop columns automatically.
+ // Or it's not merge into writes, we validate schema, and schema is compatible.
+ if (shouldValidateSchemasCompatibility) {
+ checkSchemaCompatible(latestTableSchema, canonicalizedSourceSchema, true,
+ allowAutoEvolutionColumnDrop, java.util.Collections.emptySet())
+ }
+ canonicalizedSourceSchema
+ }
+ }
+
+ /**
+ * Deducing with enabled reconciliation.
+ * Marked as Deprecated.
+ */
+ private def deduceWriterSchemaWithReconcile(sourceSchema: Schema,
+ canonicalizedSourceSchema: Schema,
+ latestTableSchema: Schema,
+ internalSchemaOpt: Option[InternalSchema],
+ opts: Map[String, String]): Schema = {
+ internalSchemaOpt match {
+ case Some(internalSchema) =>
+ // Apply schema evolution, by auto-merging write schema and read schema
+ val mergedInternalSchema = AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, internalSchema)
+ val evolvedSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema, latestTableSchema.getFullName)
+ val shouldRemoveMetaDataFromInternalSchema = sourceSchema.getFields().filter(f => f.name().equalsIgnoreCase(HoodieRecord.RECORD_KEY_METADATA_FIELD)).isEmpty
+ if (shouldRemoveMetaDataFromInternalSchema) HoodieAvroUtils.removeMetadataFields(evolvedSchema) else evolvedSchema
+ case None =>
+ // In case schema reconciliation is enabled we will employ (legacy) reconciliation
+ // strategy to produce target writer's schema (see definition below)
+ val (reconciledSchema, isCompatible) =
+ reconcileSchemasLegacy(latestTableSchema, canonicalizedSourceSchema)
+
+ // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
+ // w/ the table's one and allow schemas to diverge. This is required in cases where
+ // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
+ // only incoming dataset's projection has to match the table's schema, and not the whole one
+ val shouldValidateSchemasCompatibility = opts.getOrDefault(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key, HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.defaultValue).toBoolean
+ if (!shouldValidateSchemasCompatibility || isCompatible) {
+ reconciledSchema
} else {
- // In case reconciliation is disabled, we have to validate that the source's schema
- // is compatible w/ the table's latest schema, such that we're able to read existing table's
- // records using [[sourceSchema]].
- //
- // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
- // w/ the table's one and allow schemas to diverge. This is required in cases where
- // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
- // only incoming dataset's projection has to match the table's schema, and not the whole one
-
- if (mergeIntoWrites) {
- // if its merge into writes, do not check for projection nor schema compatibility. Writers down the line will
- // take care of it.
- canonicalizedSourceSchema
- } else {
- if (!shouldValidateSchemasCompatibility) {
- // if no validation is enabled, check for col drop
- if (allowAutoEvolutionColumnDrop) {
- canonicalizedSourceSchema
- } else {
- val reconciledSchema = if (setNullForMissingColumns) {
- AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, latestTableSchema)
- } else {
- canonicalizedSourceSchema
- }
- checkValidEvolution(reconciledSchema, latestTableSchema)
- reconciledSchema
- }
- } else {
- checkSchemaCompatible(latestTableSchema, canonicalizedSourceSchema, true,
- allowAutoEvolutionColumnDrop, java.util.Collections.emptySet())
- canonicalizedSourceSchema
- }
- }
+ log.error(
+ s"""Failed to reconcile incoming batch schema with the table's one.
+ |Incoming schema ${sourceSchema.toString(true)}
+ |Incoming schema (canonicalized) ${canonicalizedSourceSchema.toString(true)}
+ |Table's schema ${latestTableSchema.toString(true)}
+ |""".stripMargin)
+ throw new SchemaCompatibilityException("Failed to reconcile incoming schema with the table's one")
}
}
}
(hudi) 07/14: [HUDI-7491] Fixing handling null values of extra metadata in clean commit metadata (#10837)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 8e6eff945bf09803e42eb2a1e33cb515befaad05
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Fri Mar 8 01:27:06 2024 -0800
[HUDI-7491] Fixing handling null values of extra metadata in clean commit metadata (#10837)
* Fixing handling null values of extra metadata in clean commit metadata
* fixing tests
---
.../main/java/org/apache/hudi/table/action/clean/CleanPlanner.java | 4 ++--
.../src/test/java/org/apache/hudi/table/action/TestCleanPlanner.java | 2 +-
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
index 882e56b3270..b83e3ab74ea 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
@@ -234,8 +234,8 @@ public class CleanPlanner<T, I, K, O> implements Serializable {
}
private List<String> getPartitionsFromDeletedSavepoint(HoodieCleanMetadata cleanMetadata) {
- List<String> savepointedTimestampsFromLastClean = Arrays.stream(cleanMetadata.getExtraMetadata()
- .getOrDefault(SAVEPOINTED_TIMESTAMPS, StringUtils.EMPTY_STRING).split(","))
+ List<String> savepointedTimestampsFromLastClean = cleanMetadata.getExtraMetadata() == null ? Collections.emptyList()
+ : Arrays.stream(cleanMetadata.getExtraMetadata().getOrDefault(SAVEPOINTED_TIMESTAMPS, StringUtils.EMPTY_STRING).split(","))
.filter(partition -> !StringUtils.isNullOrEmpty(partition)).collect(Collectors.toList());
if (savepointedTimestampsFromLastClean.isEmpty()) {
return Collections.emptyList();
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/TestCleanPlanner.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/TestCleanPlanner.java
index 61bff2312b1..2bc1564927b 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/TestCleanPlanner.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/TestCleanPlanner.java
@@ -508,7 +508,7 @@ public class TestCleanPlanner {
extraMetadata.put(SAVEPOINTED_TIMESTAMPS, savepointsToTrack.stream().collect(Collectors.joining(",")));
}
HoodieCleanMetadata cleanMetadata = new HoodieCleanMetadata(instantTime, 100L, 10, earliestCommitToRetain, lastCompletedTime, partitionMetadata,
- CLEAN_METADATA_VERSION_2, Collections.EMPTY_MAP, extraMetadata);
+ CLEAN_METADATA_VERSION_2, Collections.EMPTY_MAP, extraMetadata.isEmpty() ? null : extraMetadata);
return Pair.of(cleanMetadata, TimelineMetadataUtils.serializeCleanMetadata(cleanMetadata));
} catch (IOException ex) {
throw new UncheckedIOException(ex);
(hudi) 09/14: [ENG-6316] Bump cleaner retention for MDT (#537) (#10655)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 866348adb39e83e7908bc0bfa6d6c1a9dc7f2a89
Author: Sagar Sumit <sa...@gmail.com>
AuthorDate: Fri Mar 8 23:47:41 2024 +0530
[ENG-6316] Bump cleaner retention for MDT (#537) (#10655)
---
.../hudi/metadata/HoodieMetadataWriteUtils.java | 28 +++++++---
.../metadata/TestHoodieMetadataWriteUtils.java | 64 ++++++++++++++++++++++
2 files changed, 84 insertions(+), 8 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
index 76fffd5d0df..48cfb46b49f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
@@ -82,6 +82,25 @@ public class HoodieMetadataWriteUtils {
String tableName = writeConfig.getTableName() + METADATA_TABLE_NAME_SUFFIX;
final long maxLogFileSizeBytes = writeConfig.getMetadataConfig().getMaxLogFileSize();
+ // Borrow the cleaner policy from the main table and adjust the cleaner policy based on the main table's cleaner policy
+ HoodieCleaningPolicy dataTableCleaningPolicy = writeConfig.getCleanerPolicy();
+ HoodieCleanConfig.Builder cleanConfigBuilder = HoodieCleanConfig.newBuilder()
+ .withAsyncClean(DEFAULT_METADATA_ASYNC_CLEAN)
+ .withAutoClean(false)
+ .withCleanerParallelism(MDT_DEFAULT_PARALLELISM)
+ .withFailedWritesCleaningPolicy(failedWritesCleaningPolicy)
+ .withCleanerPolicy(dataTableCleaningPolicy);
+
+ if (HoodieCleaningPolicy.KEEP_LATEST_COMMITS.equals(dataTableCleaningPolicy)) {
+ int retainCommits = (int) Math.max(DEFAULT_METADATA_CLEANER_COMMITS_RETAINED, writeConfig.getCleanerCommitsRetained() * 1.2);
+ cleanConfigBuilder.retainCommits(retainCommits);
+ } else if (HoodieCleaningPolicy.KEEP_LATEST_FILE_VERSIONS.equals(dataTableCleaningPolicy)) {
+ int retainFileVersions = (int) Math.ceil(writeConfig.getCleanerFileVersionsRetained() * 1.2);
+ cleanConfigBuilder.retainFileVersions(retainFileVersions);
+ } else if (HoodieCleaningPolicy.KEEP_LATEST_BY_HOURS.equals(dataTableCleaningPolicy)) {
+ int numHoursRetained = (int) Math.ceil(writeConfig.getCleanerHoursRetained() * 1.2);
+ cleanConfigBuilder.cleanerNumHoursRetained(numHoursRetained);
+ }
// Create the write config for the metadata table by borrowing options from the main write config.
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
@@ -105,14 +124,7 @@ public class HoodieMetadataWriteUtils {
.withSchema(HoodieMetadataRecord.getClassSchema().toString())
.forTable(tableName)
// we will trigger cleaning manually, to control the instant times
- .withCleanConfig(HoodieCleanConfig.newBuilder()
- .withAsyncClean(DEFAULT_METADATA_ASYNC_CLEAN)
- .withAutoClean(false)
- .withCleanerParallelism(MDT_DEFAULT_PARALLELISM)
- .withCleanerPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS)
- .withFailedWritesCleaningPolicy(failedWritesCleaningPolicy)
- .retainCommits(DEFAULT_METADATA_CLEANER_COMMITS_RETAINED)
- .build())
+ .withCleanConfig(cleanConfigBuilder.build())
// we will trigger archive manually, to ensure only regular writer invokes it
.withArchivalConfig(HoodieArchivalConfig.newBuilder()
.archiveCommitsWith(
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
new file mode 100644
index 00000000000..529d2ddfc7f
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.metadata;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieCleaningPolicy;
+import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
+import org.apache.hudi.config.HoodieCleanConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class TestHoodieMetadataWriteUtils {
+
+ @Test
+ public void testCreateMetadataWriteConfigForCleaner() {
+ HoodieWriteConfig writeConfig1 = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp")
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withCleanerPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS)
+ .retainCommits(5).build())
+ .build();
+
+ HoodieWriteConfig metadataWriteConfig1 = HoodieMetadataWriteUtils.createMetadataWriteConfig(writeConfig1, HoodieFailedWritesCleaningPolicy.EAGER);
+ assertEquals(HoodieFailedWritesCleaningPolicy.EAGER, metadataWriteConfig1.getFailedWritesCleanPolicy());
+ assertEquals(HoodieCleaningPolicy.KEEP_LATEST_COMMITS, metadataWriteConfig1.getCleanerPolicy());
+ // default value already greater than data cleaner commits retained * 1.2
+ assertEquals(HoodieMetadataConfig.DEFAULT_METADATA_CLEANER_COMMITS_RETAINED, metadataWriteConfig1.getCleanerCommitsRetained());
+
+ assertNotEquals(HoodieCleaningPolicy.KEEP_LATEST_FILE_VERSIONS, metadataWriteConfig1.getCleanerPolicy());
+ assertNotEquals(HoodieCleaningPolicy.KEEP_LATEST_BY_HOURS, metadataWriteConfig1.getCleanerPolicy());
+
+ HoodieWriteConfig writeConfig2 = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp")
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withCleanerPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS)
+ .retainCommits(20).build())
+ .build();
+ HoodieWriteConfig metadataWriteConfig2 = HoodieMetadataWriteUtils.createMetadataWriteConfig(writeConfig2, HoodieFailedWritesCleaningPolicy.EAGER);
+ assertEquals(HoodieFailedWritesCleaningPolicy.EAGER, metadataWriteConfig2.getFailedWritesCleanPolicy());
+ assertEquals(HoodieCleaningPolicy.KEEP_LATEST_COMMITS, metadataWriteConfig2.getCleanerPolicy());
+ // data cleaner commits retained * 1.2 is greater than default
+ assertEquals(24, metadataWriteConfig2.getCleanerCommitsRetained());
+ }
+}
(hudi) 14/14: [HUDI-7496] Bump mybatis from 3.4.6 to 3.5.6 in /hudi-platform-service/hudi-metaserver/hudi-metaserver-server (#7673)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 3f78130d007f1f8695d4ad200a2c04279438384c
Author: dependabot[bot] <49...@users.noreply.github.com>
AuthorDate: Sat Mar 9 12:51:28 2024 -0800
[HUDI-7496] Bump mybatis from 3.4.6 to 3.5.6 in /hudi-platform-service/hudi-metaserver/hudi-metaserver-server (#7673)
Signed-off-by: dependabot[bot] <su...@github.com>
Co-authored-by: dependabot[bot] <49...@users.noreply.github.com>
---
hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml b/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml
index 8b32f962d7c..1099dd8bf25 100644
--- a/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml
+++ b/hudi-platform-service/hudi-metaserver/hudi-metaserver-server/pom.xml
@@ -30,7 +30,7 @@
<properties>
<main.basedir>${project.parent.basedir}</main.basedir>
- <mybatis.version>3.4.6</mybatis.version>
+ <mybatis.version>3.5.6</mybatis.version>
<HikariCP.version>4.0.3</HikariCP.version>
<mysql-connector-java.version>8.0.28</mysql-connector-java.version>
</properties>