You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2022/12/14 00:03:23 UTC
[hudi] branch release-0.12.2-shadow updated (0afeb0ecc25 -> 07be18cf1ad)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a change to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
from 0afeb0ecc25 [HUDI-5353] Close file readers (#7412)
new 38e5a35e76e [HUDI-5078] Fixing isTableService for replace commits (#7037)
new 8a4c86b9582 [HUDI-5296] Allow disable schema on read after enabling (#7421)
new 16a89240b25 [HUDI-5348] Cache file slices in HoodieBackedTableMetadata (#7436)
new d30b7d3857c [HUDI-5373] Different fileids are assigned to the same bucket (#7433)
new 20956a21ac5 【HUDI-4917】Optimized the way to get HoodieBaseFile of loadColumnRangesFromFiles of Bloom Index (#6793)
new e346f181624 [HUDI-5351] Handle populateMetaFields when repartitioning in sort partitioner (#7411)
new a56c00e767f [HUDI-4113] Fix cannot parse <null> schema when use spark delete sql (#5610)
new 07be18cf1ad [HUDI-5366] Closing metadata writer from within writeClient (#7437)
The 8 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:
.../apache/hudi/client/BaseHoodieWriteClient.java | 6 +-
.../apache/hudi/index/bloom/HoodieBloomIndex.java | 13 ++-
.../org/apache/hudi/io/HoodieRangeInfoHandle.java | 8 ++
.../java/org/apache/hudi/io/HoodieReadHandle.java | 5 +
.../metadata/HoodieBackedTableMetadataWriter.java | 13 +++
.../java/org/apache/hudi/table/HoodieTable.java | 19 ++-
.../hudi/table/action/BaseActionExecutor.java | 2 +-
.../apache/hudi/client/HoodieFlinkWriteClient.java | 8 +-
.../hudi/table/HoodieFlinkCopyOnWriteTable.java | 6 -
.../hudi/table/HoodieFlinkMergeOnReadTable.java | 6 -
.../hudi/table/HoodieJavaCopyOnWriteTable.java | 6 -
.../hudi/table/HoodieJavaMergeOnReadTable.java | 6 -
.../apache/hudi/client/SparkRDDWriteClient.java | 10 +-
.../MultipleSparkJobExecutionStrategy.java | 6 +-
.../BulkInsertInternalPartitionerFactory.java | 20 ++--
...lkInsertInternalPartitionerWithRowsFactory.java | 19 +--
.../bulkinsert/GlobalSortPartitioner.java | 14 +++
.../bulkinsert/GlobalSortPartitionerWithRows.java | 14 +++
...PartitionPathRepartitionAndSortPartitioner.java | 12 +-
...nPathRepartitionAndSortPartitionerWithRows.java | 12 +-
.../PartitionPathRepartitionPartitioner.java | 12 +-
...artitionPathRepartitionPartitionerWithRows.java | 12 +-
.../PartitionSortPartitionerWithRows.java | 14 +++
.../bulkinsert/RDDPartitionSortPartitioner.java | 14 +++
.../SparkHoodieBackedTableMetadataWriter.java | 1 +
.../hudi/table/HoodieSparkCopyOnWriteTable.java | 5 -
.../hudi/table/HoodieSparkMergeOnReadTable.java | 5 -
.../TestBulkInsertInternalPartitioner.java | 83 ++++++++-----
.../TestBulkInsertInternalPartitionerForRows.java | 69 +++++++----
.../apache/hudi/io/TestHoodieTimelineArchiver.java | 23 ++++
.../hudi/common/table/timeline/HoodieInstant.java | 2 +-
.../hudi/metadata/HoodieBackedTableMetadata.java | 13 ++-
.../hudi/metadata/HoodieTableMetadataUtil.java | 10 +-
.../table/timeline/TestHoodieActiveTimeline.java | 2 +-
.../sink/bucket/BucketStreamWriteFunction.java | 2 +-
.../scala/org/apache/hudi/HoodieBaseRelation.scala | 20 ++--
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 129 ++++++++++++++-------
.../org/apache/hudi/IncrementalRelation.scala | 6 +-
.../sql/hudi/TestAlterTableDropPartition.scala | 62 +++++++++-
.../org/apache/spark/sql/hudi/TestSpark3DDL.scala | 54 ++++++++-
.../apache/hudi/utilities/TestHoodieIndexer.java | 7 +-
41 files changed, 553 insertions(+), 197 deletions(-)
[hudi] 08/08: [HUDI-5366] Closing metadata writer from within writeClient (#7437)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 07be18cf1add17bf90010104a019a80f9530ccc2
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Tue Dec 13 08:12:47 2022 -0800
[HUDI-5366] Closing metadata writer from within writeClient (#7437)
* Closing metadata writer from within writeClient
* Close metadata writer in flink client
Co-authored-by: Sagar Sumit <sa...@gmail.com>
---
.../hudi/metadata/HoodieBackedTableMetadataWriter.java | 13 +++++++++++++
.../java/org/apache/hudi/client/HoodieFlinkWriteClient.java | 6 ++++++
.../java/org/apache/hudi/client/SparkRDDWriteClient.java | 8 +++++++-
.../hudi/metadata/SparkHoodieBackedTableMetadataWriter.java | 1 +
4 files changed, 27 insertions(+), 1 deletion(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
index 42e35cfa30f..9d758052f71 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
@@ -751,6 +751,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
LOG.warn("Deleting pending indexing instant from the timeline for partition: " + partitionPath);
deletePendingIndexingInstant(dataMetaClient, partitionPath);
}
+ closeInternal();
}
/**
@@ -884,6 +885,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
public void update(HoodieCommitMetadata commitMetadata, String instantTime, boolean isTableServiceAction) {
processAndCommit(instantTime, () -> HoodieTableMetadataUtil.convertMetadataToRecords(
engineContext, commitMetadata, instantTime, getRecordsGenerationParams()), !isTableServiceAction);
+ closeInternal();
}
/**
@@ -896,6 +898,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
public void update(HoodieCleanMetadata cleanMetadata, String instantTime) {
processAndCommit(instantTime, () -> HoodieTableMetadataUtil.convertMetadataToRecords(engineContext,
cleanMetadata, getRecordsGenerationParams(), instantTime), false);
+ closeInternal();
}
/**
@@ -909,6 +912,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
processAndCommit(instantTime, () -> HoodieTableMetadataUtil.convertMetadataToRecords(engineContext,
metadataMetaClient.getActiveTimeline(), restoreMetadata, getRecordsGenerationParams(), instantTime,
metadata.getSyncedInstantTime()), false);
+ closeInternal();
}
/**
@@ -937,6 +941,7 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
rollbackMetadata, getRecordsGenerationParams(), instantTime,
metadata.getSyncedInstantTime(), wasSynced);
commit(instantTime, records, false);
+ closeInternal();
}
}
@@ -1124,6 +1129,14 @@ public abstract class HoodieBackedTableMetadataWriter implements HoodieTableMeta
return filesPartitionRecords.union(fileListRecords);
}
+ protected void closeInternal() {
+ try {
+ close();
+ } catch (Exception e) {
+ throw new HoodieException("Failed to close HoodieMetadata writer ", e);
+ }
+ }
+
/**
* A class which represents a directory and the files and directories inside it.
* <p>
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
index b4003712c56..5430dc6eecc 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -43,6 +43,7 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieClusteringException;
import org.apache.hudi.exception.HoodieCommitException;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.index.FlinkHoodieIndexFactory;
import org.apache.hudi.index.HoodieIndex;
@@ -290,6 +291,11 @@ public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
// jobs.
this.metadataWriter.initTableMetadata();
this.metadataWriter.update(metadata, instantTime, getHoodieTable().isTableServiceAction(actionType, instantTime));
+ try {
+ this.metadataWriter.close();
+ } catch (Exception e) {
+ throw new HoodieException("Failed to close metadata writer ", e);
+ }
}
/**
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
index fa568a7a6fe..fe3a0063567 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -45,6 +45,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaRDD;
import org.apache.hudi.exception.HoodieClusteringException;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieWriteConflictException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.index.SparkHoodieIndexFactory;
@@ -453,8 +454,13 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
*/
private void initializeMetadataTable(Option<String> inFlightInstantTimestamp) {
if (config.isMetadataTableEnabled()) {
- SparkHoodieBackedTableMetadataWriter.create(context.getHadoopConf().get(), config,
+ HoodieTableMetadataWriter writer = SparkHoodieBackedTableMetadataWriter.create(context.getHadoopConf().get(), config,
context, Option.empty(), inFlightInstantTimestamp);
+ try {
+ writer.close();
+ } catch (Exception e) {
+ throw new HoodieException("Failed to instantiate Metadata table ", e);
+ }
}
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
index 7d94b2d4f53..272d3d47985 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java
@@ -192,5 +192,6 @@ public class SparkHoodieBackedTableMetadataWriter extends HoodieBackedTableMetad
writeClient.startCommitWithTime(instantTime, actionType);
writeClient.deletePartitions(partitionsToDrop, instantTime);
}
+ closeInternal();
}
}
[hudi] 06/08: [HUDI-5351] Handle populateMetaFields when repartitioning in sort partitioner (#7411)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e346f18162426646e12d40958051af49f081a30b
Author: Sagar Sumit <sa...@gmail.com>
AuthorDate: Tue Dec 13 10:22:10 2022 +0530
[HUDI-5351] Handle populateMetaFields when repartitioning in sort partitioner (#7411)
---
.../MultipleSparkJobExecutionStrategy.java | 6 +-
.../BulkInsertInternalPartitionerFactory.java | 20 +++---
...lkInsertInternalPartitionerWithRowsFactory.java | 19 ++---
.../bulkinsert/GlobalSortPartitioner.java | 14 ++++
.../bulkinsert/GlobalSortPartitionerWithRows.java | 14 ++++
...PartitionPathRepartitionAndSortPartitioner.java | 12 +++-
...nPathRepartitionAndSortPartitionerWithRows.java | 12 +++-
.../PartitionPathRepartitionPartitioner.java | 12 +++-
...artitionPathRepartitionPartitionerWithRows.java | 12 +++-
.../PartitionSortPartitionerWithRows.java | 14 ++++
.../bulkinsert/RDDPartitionSortPartitioner.java | 14 ++++
.../TestBulkInsertInternalPartitioner.java | 83 ++++++++++++++--------
.../TestBulkInsertInternalPartitionerForRows.java | 69 ++++++++++++------
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 5 +-
14 files changed, 223 insertions(+), 83 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
index e37eb4316d0..275f954781e 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
@@ -206,10 +206,8 @@ public abstract class MultipleSparkJobExecutionStrategy<T extends HoodieRecordPa
throw new UnsupportedOperationException(String.format("Layout optimization strategy '%s' is not supported", layoutOptStrategy));
}
}).orElse(isRowPartitioner
- ? BulkInsertInternalPartitionerWithRowsFactory.get(
- getWriteConfig().getBulkInsertSortMode(), getHoodieTable().isPartitioned(), true)
- : BulkInsertInternalPartitionerFactory.get(
- getWriteConfig().getBulkInsertSortMode(), getHoodieTable().isPartitioned(), true));
+ ? BulkInsertInternalPartitionerWithRowsFactory.get(getWriteConfig(), getHoodieTable().isPartitioned(), true)
+ : BulkInsertInternalPartitionerFactory.get(getHoodieTable(), getWriteConfig(), true));
}
/**
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java
index 900d2729f10..c854d72c5f9 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java
@@ -30,32 +30,30 @@ import org.apache.hudi.table.HoodieTable;
public abstract class BulkInsertInternalPartitionerFactory {
public static BulkInsertPartitioner get(HoodieTable table, HoodieWriteConfig config) {
- return get(config.getBulkInsertSortMode(), table.isPartitioned(), false);
+ return get(table, config, false);
}
public static BulkInsertPartitioner get(
HoodieTable table, HoodieWriteConfig config, boolean enforceNumOutputPartitions) {
- return get(config.getBulkInsertSortMode(), table.isPartitioned(), enforceNumOutputPartitions);
+ return get(config, table.isPartitioned(), enforceNumOutputPartitions);
}
- public static BulkInsertPartitioner get(BulkInsertSortMode sortMode, boolean isTablePartitioned) {
- return get(sortMode, isTablePartitioned, false);
- }
-
- public static BulkInsertPartitioner get(BulkInsertSortMode sortMode,
+ public static BulkInsertPartitioner get(HoodieWriteConfig config,
boolean isTablePartitioned,
boolean enforceNumOutputPartitions) {
+ BulkInsertSortMode sortMode = config.getBulkInsertSortMode();
+
switch (sortMode) {
case NONE:
return new NonSortPartitioner(enforceNumOutputPartitions);
case GLOBAL_SORT:
- return new GlobalSortPartitioner();
+ return new GlobalSortPartitioner(config);
case PARTITION_SORT:
- return new RDDPartitionSortPartitioner();
+ return new RDDPartitionSortPartitioner(config);
case PARTITION_PATH_REPARTITION:
- return new PartitionPathRepartitionPartitioner(isTablePartitioned);
+ return new PartitionPathRepartitionPartitioner(isTablePartitioned, config);
case PARTITION_PATH_REPARTITION_AND_SORT:
- return new PartitionPathRepartitionAndSortPartitioner(isTablePartitioned);
+ return new PartitionPathRepartitionAndSortPartitioner(isTablePartitioned, config);
default:
throw new HoodieException("The bulk insert sort mode \"" + sortMode.name() + "\" is not supported.");
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java
index 218eae0dc94..07995e50d6a 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java
@@ -18,6 +18,7 @@
package org.apache.hudi.execution.bulkinsert;
+import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.sql.Dataset;
@@ -29,24 +30,26 @@ import org.apache.spark.sql.Row;
*/
public abstract class BulkInsertInternalPartitionerWithRowsFactory {
- public static BulkInsertPartitioner<Dataset<Row>> get(BulkInsertSortMode sortMode,
+ public static BulkInsertPartitioner<Dataset<Row>> get(HoodieWriteConfig config,
boolean isTablePartitioned) {
- return get(sortMode, isTablePartitioned, false);
+ return get(config, isTablePartitioned, false);
}
- public static BulkInsertPartitioner<Dataset<Row>> get(
- BulkInsertSortMode sortMode, boolean isTablePartitioned, boolean enforceNumOutputPartitions) {
+ public static BulkInsertPartitioner<Dataset<Row>> get(HoodieWriteConfig config,
+ boolean isTablePartitioned,
+ boolean enforceNumOutputPartitions) {
+ BulkInsertSortMode sortMode = config.getBulkInsertSortMode();
switch (sortMode) {
case NONE:
return new NonSortPartitionerWithRows(enforceNumOutputPartitions);
case GLOBAL_SORT:
- return new GlobalSortPartitionerWithRows();
+ return new GlobalSortPartitionerWithRows(config);
case PARTITION_SORT:
- return new PartitionSortPartitionerWithRows();
+ return new PartitionSortPartitionerWithRows(config);
case PARTITION_PATH_REPARTITION:
- return new PartitionPathRepartitionPartitionerWithRows(isTablePartitioned);
+ return new PartitionPathRepartitionPartitionerWithRows(isTablePartitioned, config);
case PARTITION_PATH_REPARTITION_AND_SORT:
- return new PartitionPathRepartitionAndSortPartitionerWithRows(isTablePartitioned);
+ return new PartitionPathRepartitionAndSortPartitionerWithRows(isTablePartitioned, config);
default:
throw new UnsupportedOperationException("The bulk insert sort mode \"" + sortMode.name() + "\" is not supported.");
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitioner.java
index a184c009a1b..e10d23743da 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitioner.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitioner.java
@@ -20,10 +20,14 @@ package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.api.java.JavaRDD;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.GLOBAL_SORT;
+
/**
* A built-in partitioner that does global sorting for the input records across partitions
* after repartition for bulk insert operation, corresponding to the
@@ -34,9 +38,19 @@ import org.apache.spark.api.java.JavaRDD;
public class GlobalSortPartitioner<T extends HoodieRecordPayload>
implements BulkInsertPartitioner<JavaRDD<HoodieRecord<T>>> {
+ private final boolean shouldPopulateMetaFields;
+
+ public GlobalSortPartitioner(HoodieWriteConfig config) {
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
@Override
public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records,
int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(GLOBAL_SORT.name() + " mode requires meta-fields to be enabled");
+ }
+
// Now, sort the records and line them up nicely for loading.
return records.sortBy(record -> {
// Let's use "partitionPath + key" as the sort key. Spark, will ensure
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitionerWithRows.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitionerWithRows.java
index 24bcc0aff0d..1a8e87a3823 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitionerWithRows.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/GlobalSortPartitionerWithRows.java
@@ -19,19 +19,33 @@
package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.GLOBAL_SORT;
+
/**
* A built-in partitioner that does global sorting for the input Rows across partitions after repartition for bulk insert operation, corresponding to the {@code BulkInsertSortMode.GLOBAL_SORT} mode.
*/
public class GlobalSortPartitionerWithRows implements BulkInsertPartitioner<Dataset<Row>> {
+ private final boolean shouldPopulateMetaFields;
+
+ public GlobalSortPartitionerWithRows(HoodieWriteConfig config) {
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
@Override
public Dataset<Row> repartitionRecords(Dataset<Row> rows, int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(GLOBAL_SORT.name() + " mode requires meta-fields to be enabled");
+ }
+
// Now, sort the records and line them up nicely for loading.
// Let's use "partitionPath + key" as the sort key.
return rows.sort(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD), functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD))
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitioner.java
index e8e1e2072f5..a47f1f9df43 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitioner.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitioner.java
@@ -21,12 +21,16 @@ package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.api.java.JavaRDD;
import scala.Tuple2;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT;
+
/**
* A built-in partitioner that does the following for input records for bulk insert operation
* <p>
@@ -45,14 +49,20 @@ public class PartitionPathRepartitionAndSortPartitioner<T extends HoodieRecordPa
implements BulkInsertPartitioner<JavaRDD<HoodieRecord<T>>> {
private final boolean isTablePartitioned;
+ private final boolean shouldPopulateMetaFields;
- public PartitionPathRepartitionAndSortPartitioner(boolean isTablePartitioned) {
+ public PartitionPathRepartitionAndSortPartitioner(boolean isTablePartitioned, HoodieWriteConfig config) {
this.isTablePartitioned = isTablePartitioned;
+ this.shouldPopulateMetaFields = config.populateMetaFields();
}
@Override
public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records,
int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(PARTITION_PATH_REPARTITION_AND_SORT.name() + " mode requires meta-fields to be enabled");
+ }
+
if (isTablePartitioned) {
PartitionPathRDDPartitioner partitioner = new PartitionPathRDDPartitioner(
(partitionPath) -> (String) partitionPath, outputSparkPartitions);
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitionerWithRows.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitionerWithRows.java
index cf3ff1acfa4..ff505624913 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitionerWithRows.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionAndSortPartitionerWithRows.java
@@ -20,12 +20,16 @@
package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT;
+
/**
* A built-in partitioner that does the following for input rows for bulk insert operation
* <p>
@@ -41,13 +45,19 @@ import org.apache.spark.sql.Row;
public class PartitionPathRepartitionAndSortPartitionerWithRows implements BulkInsertPartitioner<Dataset<Row>> {
private final boolean isTablePartitioned;
+ private final boolean shouldPopulateMetaFields;
- public PartitionPathRepartitionAndSortPartitionerWithRows(boolean isTablePartitioned) {
+ public PartitionPathRepartitionAndSortPartitionerWithRows(boolean isTablePartitioned, HoodieWriteConfig config) {
this.isTablePartitioned = isTablePartitioned;
+ this.shouldPopulateMetaFields = config.populateMetaFields();
}
@Override
public Dataset<Row> repartitionRecords(Dataset<Row> rows, int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(PARTITION_PATH_REPARTITION_AND_SORT.name() + " mode requires meta-fields to be enabled");
+ }
+
if (isTablePartitioned) {
return rows.repartition(outputSparkPartitions, new Column(HoodieRecord.PARTITION_PATH_METADATA_FIELD))
.sortWithinPartitions(new Column(HoodieRecord.PARTITION_PATH_METADATA_FIELD));
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitioner.java
index 5931b565757..393e47a61e3 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitioner.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitioner.java
@@ -21,12 +21,16 @@ package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.api.java.JavaRDD;
import scala.Tuple2;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_PATH_REPARTITION;
+
/**
* A built-in partitioner that does the following for input records for bulk insert operation
* <p>
@@ -44,14 +48,20 @@ public class PartitionPathRepartitionPartitioner<T extends HoodieRecordPayload>
implements BulkInsertPartitioner<JavaRDD<HoodieRecord<T>>> {
private final boolean isTablePartitioned;
+ private final boolean shouldPopulateMetaFields;
- public PartitionPathRepartitionPartitioner(boolean isTablePartitioned) {
+ public PartitionPathRepartitionPartitioner(boolean isTablePartitioned, HoodieWriteConfig config) {
this.isTablePartitioned = isTablePartitioned;
+ this.shouldPopulateMetaFields = config.populateMetaFields();
}
@Override
public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records,
int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(PARTITION_PATH_REPARTITION.name() + " mode requires meta-fields to be enabled");
+ }
+
if (isTablePartitioned) {
PartitionPathRDDPartitioner partitioner = new PartitionPathRDDPartitioner(
(partitionPath) -> (String) partitionPath, outputSparkPartitions);
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitionerWithRows.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitionerWithRows.java
index 62d9edbca84..966648d579c 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitionerWithRows.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRepartitionPartitionerWithRows.java
@@ -20,12 +20,16 @@
package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_PATH_REPARTITION;
+
/**
* A built-in partitioner that does the following for input rows for bulk insert operation
* <p>
@@ -40,13 +44,19 @@ import org.apache.spark.sql.Row;
public class PartitionPathRepartitionPartitionerWithRows implements BulkInsertPartitioner<Dataset<Row>> {
private final boolean isTablePartitioned;
+ private final boolean shouldPopulateMetaFields;
- public PartitionPathRepartitionPartitionerWithRows(boolean isTablePartitioned) {
+ public PartitionPathRepartitionPartitionerWithRows(boolean isTablePartitioned, HoodieWriteConfig config) {
this.isTablePartitioned = isTablePartitioned;
+ this.shouldPopulateMetaFields = config.populateMetaFields();
}
@Override
public Dataset<Row> repartitionRecords(Dataset<Row> rows, int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(PARTITION_PATH_REPARTITION.name() + " mode requires meta-fields to be enabled");
+ }
+
if (isTablePartitioned) {
return rows.repartition(outputSparkPartitions, new Column(HoodieRecord.PARTITION_PATH_METADATA_FIELD));
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionSortPartitionerWithRows.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionSortPartitionerWithRows.java
index b669c338f86..c5c21f14a8e 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionSortPartitionerWithRows.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionSortPartitionerWithRows.java
@@ -19,18 +19,32 @@
package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_SORT;
+
/**
* A built-in partitioner that does local sorting for each spark partitions after coalesce for bulk insert operation, corresponding to the {@code BulkInsertSortMode.PARTITION_SORT} mode.
*/
public class PartitionSortPartitionerWithRows implements BulkInsertPartitioner<Dataset<Row>> {
+ private final boolean shouldPopulateMetaFields;
+
+ public PartitionSortPartitionerWithRows(HoodieWriteConfig config) {
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
@Override
public Dataset<Row> repartitionRecords(Dataset<Row> rows, int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(PARTITION_SORT.name() + " mode requires meta-fields to be enabled");
+ }
+
return rows.coalesce(outputSparkPartitions).sortWithinPartitions(HoodieRecord.PARTITION_PATH_METADATA_FIELD, HoodieRecord.RECORD_KEY_METADATA_FIELD);
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDPartitionSortPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDPartitionSortPartitioner.java
index 9526ad58564..b5dc83cc2fa 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDPartitionSortPartitioner.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDPartitionSortPartitioner.java
@@ -20,6 +20,8 @@ package org.apache.hudi.execution.bulkinsert;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.api.java.JavaRDD;
@@ -30,6 +32,8 @@ import java.util.List;
import scala.Tuple2;
+import static org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_SORT;
+
/**
* A built-in partitioner that does local sorting for each RDD partition
* after coalesce for bulk insert operation, corresponding to the
@@ -40,9 +44,19 @@ import scala.Tuple2;
public class RDDPartitionSortPartitioner<T extends HoodieRecordPayload>
implements BulkInsertPartitioner<JavaRDD<HoodieRecord<T>>> {
+ private final boolean shouldPopulateMetaFields;
+
+ public RDDPartitionSortPartitioner(HoodieWriteConfig config) {
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
@Override
public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records,
int outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(PARTITION_SORT.name() + " mode requires meta-fields to be enabled");
+ }
+
return records.coalesce(outputSparkPartitions)
.mapToPair(record ->
new Tuple2<>(
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitioner.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitioner.java
index 7bc64b54457..11420c180eb 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitioner.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitioner.java
@@ -23,6 +23,7 @@ import org.apache.hudi.common.model.HoodieRecordPayload;
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.HoodieIOException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.testutils.HoodieClientTestBase;
@@ -48,6 +49,7 @@ import java.util.stream.Stream;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
public class TestBulkInsertInternalPartitioner extends HoodieClientTestBase implements Serializable {
private static final Comparator<HoodieRecord<? extends HoodieRecordPayload>> KEY_COMPARATOR =
@@ -83,16 +85,21 @@ public class TestBulkInsertInternalPartitioner extends HoodieClientTestBase impl
// boolean isTablePartitioned,
// boolean enforceNumOutputPartitions,
// boolean isGloballySorted,
- // boolean isLocallySorted
+ // boolean isLocallySorted,
+ // boolean populateMetaFields
Object[][] data = new Object[][] {
- {BulkInsertSortMode.GLOBAL_SORT, true, true, true, true},
- {BulkInsertSortMode.PARTITION_SORT, true, true, false, true},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION, true, true, false, false},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION, false, true, false, false},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, true, true, false, false},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, false, true, false, false},
- {BulkInsertSortMode.NONE, true, true, false, false},
- {BulkInsertSortMode.NONE, true, false, false, false}
+ {BulkInsertSortMode.GLOBAL_SORT, true, true, true, true, true},
+ {BulkInsertSortMode.PARTITION_SORT, true, true, false, true, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION, true, true, false, false, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION, false, true, false, false, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, true, true, false, false, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, false, true, false, false, true},
+ {BulkInsertSortMode.NONE, true, true, false, false, true},
+ {BulkInsertSortMode.NONE, true, false, false, false, true},
+ {BulkInsertSortMode.GLOBAL_SORT, true, true, true, true, false},
+ {BulkInsertSortMode.PARTITION_SORT, true, true, false, true, false},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION, true, true, false, false, false},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, true, true, false, false, false}
};
return Stream.of(data).map(Arguments::of);
}
@@ -109,7 +116,8 @@ public class TestBulkInsertInternalPartitioner extends HoodieClientTestBase impl
boolean enforceNumOutputPartitions,
boolean isGloballySorted,
boolean isLocallySorted,
- Map<String, Long> expectedPartitionNumRecords) {
+ Map<String, Long> expectedPartitionNumRecords,
+ boolean populateMetaFields) {
testBulkInsertInternalPartitioner(
partitioner,
records,
@@ -117,7 +125,8 @@ public class TestBulkInsertInternalPartitioner extends HoodieClientTestBase impl
isGloballySorted,
isLocallySorted,
expectedPartitionNumRecords,
- Option.empty());
+ Option.empty(),
+ populateMetaFields);
}
private void testBulkInsertInternalPartitioner(BulkInsertPartitioner partitioner,
@@ -126,8 +135,13 @@ public class TestBulkInsertInternalPartitioner extends HoodieClientTestBase impl
boolean isGloballySorted,
boolean isLocallySorted,
Map<String, Long> expectedPartitionNumRecords,
- Option<Comparator<HoodieRecord<? extends HoodieRecordPayload>>> comparator) {
+ Option<Comparator<HoodieRecord<? extends HoodieRecordPayload>>> comparator,
+ boolean populateMetaFields) {
int numPartitions = 2;
+ if (!populateMetaFields) {
+ assertThrows(HoodieException.class, () -> partitioner.repartitionRecords(records, numPartitions));
+ return;
+ }
JavaRDD<HoodieRecord<? extends HoodieRecordPayload>> actualRecords =
(JavaRDD<HoodieRecord<? extends HoodieRecordPayload>>) partitioner.repartitionRecords(records, numPartitions);
assertEquals(
@@ -163,25 +177,35 @@ public class TestBulkInsertInternalPartitioner extends HoodieClientTestBase impl
boolean isTablePartitioned,
boolean enforceNumOutputPartitions,
boolean isGloballySorted,
- boolean isLocallySorted) {
+ boolean isLocallySorted,
+ boolean populateMetaFields) {
JavaRDD<HoodieRecord> records1 = generateTestRecordsForBulkInsert(jsc);
JavaRDD<HoodieRecord> records2 = generateTripleTestRecordsForBulkInsert(jsc);
+
+ HoodieWriteConfig config = HoodieWriteConfig
+ .newBuilder()
+ .withPath("/")
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withBulkInsertSortMode(sortMode.name())
+ .withPopulateMetaFields(populateMetaFields)
+ .build();
+
testBulkInsertInternalPartitioner(
- BulkInsertInternalPartitionerFactory.get(
- sortMode, isTablePartitioned, enforceNumOutputPartitions),
+ BulkInsertInternalPartitionerFactory.get(config, isTablePartitioned, enforceNumOutputPartitions),
records1,
enforceNumOutputPartitions,
isGloballySorted,
isLocallySorted,
- generateExpectedPartitionNumRecords(records1));
+ generateExpectedPartitionNumRecords(records1),
+ populateMetaFields);
testBulkInsertInternalPartitioner(
- BulkInsertInternalPartitionerFactory.get(
- sortMode, isTablePartitioned, enforceNumOutputPartitions),
+ BulkInsertInternalPartitionerFactory.get(config, isTablePartitioned, enforceNumOutputPartitions),
records2,
enforceNumOutputPartitions,
isGloballySorted,
isLocallySorted,
- generateExpectedPartitionNumRecords(records2));
+ generateExpectedPartitionNumRecords(records2),
+ populateMetaFields);
}
@Test
@@ -193,22 +217,21 @@ public class TestBulkInsertInternalPartitioner extends HoodieClientTestBase impl
JavaRDD<HoodieRecord> records1 = generateTestRecordsForBulkInsert(jsc);
JavaRDD<HoodieRecord> records2 = generateTripleTestRecordsForBulkInsert(jsc);
testBulkInsertInternalPartitioner(new RDDCustomColumnsSortPartitioner(sortColumns, HoodieTestDataGenerator.AVRO_SCHEMA, false),
- records1, true, true, true, generateExpectedPartitionNumRecords(records1), Option.of(columnComparator));
+ records1, true, true, true, generateExpectedPartitionNumRecords(records1), Option.of(columnComparator), true);
testBulkInsertInternalPartitioner(new RDDCustomColumnsSortPartitioner(sortColumns, HoodieTestDataGenerator.AVRO_SCHEMA, false),
- records2, true, true, true, generateExpectedPartitionNumRecords(records2), Option.of(columnComparator));
+ records2, true, true, true, generateExpectedPartitionNumRecords(records2), Option.of(columnComparator), true);
HoodieWriteConfig config = HoodieWriteConfig
- .newBuilder()
- .withPath("/")
- .withSchema(TRIP_EXAMPLE_SCHEMA)
- .withUserDefinedBulkInsertPartitionerClass(RDDCustomColumnsSortPartitioner.class.getName())
- .withUserDefinedBulkInsertPartitionerSortColumns(sortColumnString)
- .build();
+ .newBuilder()
+ .withPath("/")
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withUserDefinedBulkInsertPartitionerClass(RDDCustomColumnsSortPartitioner.class.getName())
+ .withUserDefinedBulkInsertPartitionerSortColumns(sortColumnString)
+ .build();
testBulkInsertInternalPartitioner(new RDDCustomColumnsSortPartitioner(config),
- records1, true, true, true, generateExpectedPartitionNumRecords(records1), Option.of(columnComparator));
+ records1, true, true, true, generateExpectedPartitionNumRecords(records1), Option.of(columnComparator), true);
testBulkInsertInternalPartitioner(new RDDCustomColumnsSortPartitioner(config),
- records2, true, true, true, generateExpectedPartitionNumRecords(records2), Option.of(columnComparator));
-
+ records2, true, true, true, generateExpectedPartitionNumRecords(records2), Option.of(columnComparator), true);
}
private Comparator<HoodieRecord<? extends HoodieRecordPayload>> getCustomColumnComparator(Schema schema, String[] sortColumns) {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitionerForRows.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitionerForRows.java
index de827f7a450..6332e00ba67 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitionerForRows.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestBulkInsertInternalPartitionerForRows.java
@@ -22,6 +22,7 @@ import org.apache.hudi.common.model.HoodieRecord;
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.table.BulkInsertPartitioner;
import org.apache.hudi.testutils.HoodieClientTestHarness;
import org.apache.hudi.testutils.SparkDatasetTestUtils;
@@ -44,7 +45,9 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
+import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
/**
* Unit tests {@link BulkInsertPartitioner}s with Rows.
@@ -53,6 +56,7 @@ public class TestBulkInsertInternalPartitionerForRows extends HoodieClientTestHa
private static final Comparator<Row> KEY_COMPARATOR =
Comparator.comparing(o -> (o.getAs(HoodieRecord.PARTITION_PATH_METADATA_FIELD) + "+" + o.getAs(HoodieRecord.RECORD_KEY_METADATA_FIELD)));
+
@BeforeEach
public void setUp() throws Exception {
initSparkContexts("TestBulkInsertInternalPartitionerForRows");
@@ -71,16 +75,21 @@ public class TestBulkInsertInternalPartitionerForRows extends HoodieClientTestHa
// boolean isTablePartitioned,
// boolean enforceNumOutputPartitions,
// boolean isGloballySorted,
- // boolean isLocallySorted
+ // boolean isLocallySorted,
+ // boolean populateMetaFields
Object[][] data = new Object[][] {
- {BulkInsertSortMode.GLOBAL_SORT, true, true, true, true},
- {BulkInsertSortMode.PARTITION_SORT, true, true, false, true},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION, true, true, false, false},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION, false, true, false, false},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, true, true, false, false},
- {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, false, true, false, false},
- {BulkInsertSortMode.NONE, true, true, false, false},
- {BulkInsertSortMode.NONE, true, false, false, false}
+ {BulkInsertSortMode.GLOBAL_SORT, true, true, true, true, true},
+ {BulkInsertSortMode.PARTITION_SORT, true, true, false, true, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION, true, true, false, false, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION, false, true, false, false, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, true, true, false, false, true},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, false, true, false, false, true},
+ {BulkInsertSortMode.NONE, true, true, false, false, true},
+ {BulkInsertSortMode.NONE, true, false, false, false, true},
+ {BulkInsertSortMode.GLOBAL_SORT, true, true, true, true, false},
+ {BulkInsertSortMode.PARTITION_SORT, true, true, false, true, false},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION, true, true, false, false, false},
+ {BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT, true, true, false, false, false, false},
};
return Stream.of(data).map(Arguments::of);
}
@@ -91,28 +100,37 @@ public class TestBulkInsertInternalPartitionerForRows extends HoodieClientTestHa
boolean isTablePartitioned,
boolean enforceNumOutputPartitions,
boolean isGloballySorted,
- boolean isLocallySorted)
- throws Exception {
+ boolean isLocallySorted,
+ boolean populateMetaFields) {
Dataset<Row> records1 = generateTestRecords();
Dataset<Row> records2 = generateTestRecords();
+
+ HoodieWriteConfig config = HoodieWriteConfig
+ .newBuilder()
+ .withPath("/")
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withBulkInsertSortMode(sortMode.name())
+ .withPopulateMetaFields(populateMetaFields)
+ .build();
+
testBulkInsertInternalPartitioner(
- BulkInsertInternalPartitionerWithRowsFactory.get(
- sortMode, isTablePartitioned, enforceNumOutputPartitions),
+ BulkInsertInternalPartitionerWithRowsFactory.get(config, isTablePartitioned, enforceNumOutputPartitions),
records1,
enforceNumOutputPartitions,
isGloballySorted,
isLocallySorted,
generateExpectedPartitionNumRecords(records1),
- Option.empty());
+ Option.empty(),
+ populateMetaFields);
testBulkInsertInternalPartitioner(
- BulkInsertInternalPartitionerWithRowsFactory.get(
- sortMode, isTablePartitioned, enforceNumOutputPartitions),
+ BulkInsertInternalPartitionerWithRowsFactory.get(config, isTablePartitioned, enforceNumOutputPartitions),
records2,
enforceNumOutputPartitions,
isGloballySorted,
isLocallySorted,
generateExpectedPartitionNumRecords(records2),
- Option.empty());
+ Option.empty(),
+ populateMetaFields);
}
@Test
@@ -124,9 +142,9 @@ public class TestBulkInsertInternalPartitionerForRows extends HoodieClientTestHa
Comparator<Row> comparator = getCustomColumnComparator(sortColumns);
testBulkInsertInternalPartitioner(new RowCustomColumnsSortPartitioner(sortColumns),
- records1, true, false, true, generateExpectedPartitionNumRecords(records1), Option.of(comparator));
+ records1, true, false, true, generateExpectedPartitionNumRecords(records1), Option.of(comparator), true);
testBulkInsertInternalPartitioner(new RowCustomColumnsSortPartitioner(sortColumns),
- records2, true, false, true, generateExpectedPartitionNumRecords(records2), Option.of(comparator));
+ records2, true, false, true, generateExpectedPartitionNumRecords(records2), Option.of(comparator), true);
HoodieWriteConfig config = HoodieWriteConfig
.newBuilder()
@@ -135,9 +153,9 @@ public class TestBulkInsertInternalPartitionerForRows extends HoodieClientTestHa
.withUserDefinedBulkInsertPartitionerSortColumns(sortColumnString)
.build();
testBulkInsertInternalPartitioner(new RowCustomColumnsSortPartitioner(config),
- records1, true, false, true, generateExpectedPartitionNumRecords(records1), Option.of(comparator));
+ records1, true, false, true, generateExpectedPartitionNumRecords(records1), Option.of(comparator), true);
testBulkInsertInternalPartitioner(new RowCustomColumnsSortPartitioner(config),
- records2, true, false, true, generateExpectedPartitionNumRecords(records2), Option.of(comparator));
+ records2, true, false, true, generateExpectedPartitionNumRecords(records2), Option.of(comparator), true);
}
private void testBulkInsertInternalPartitioner(BulkInsertPartitioner partitioner,
@@ -146,8 +164,13 @@ public class TestBulkInsertInternalPartitionerForRows extends HoodieClientTestHa
boolean isGloballySorted,
boolean isLocallySorted,
Map<String, Long> expectedPartitionNumRecords,
- Option<Comparator<Row>> comparator) {
+ Option<Comparator<Row>> comparator,
+ boolean populateMetaFields) {
int numPartitions = 2;
+ if (!populateMetaFields) {
+ assertThrows(HoodieException.class, () -> partitioner.repartitionRecords(rows, numPartitions));
+ return;
+ }
Dataset<Row> actualRecords = (Dataset<Row>) partitioner.repartitionRecords(rows, numPartitions);
assertEquals(
enforceNumOutputPartitions ? numPartitions : rows.rdd().getNumPartitions(),
@@ -198,7 +221,7 @@ public class TestBulkInsertInternalPartitionerForRows extends HoodieClientTestHa
private void verifyRowsAscendingOrder(List<Row> records, Option<Comparator<Row>> comparator) {
List<Row> expectedRecords = new ArrayList<>(records);
- Collections.sort(expectedRecords,comparator.orElse(KEY_COMPARATOR));
+ Collections.sort(expectedRecords, comparator.orElse(KEY_COMPARATOR));
assertEquals(expectedRecords, records);
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index 003dbd35e11..d8f9ab8435b 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -639,8 +639,7 @@ object HoodieSparkSqlWriter {
if (userDefinedBulkInsertPartitionerOpt.isPresent) {
userDefinedBulkInsertPartitionerOpt.get
} else {
- BulkInsertInternalPartitionerWithRowsFactory.get(
- writeConfig.getBulkInsertSortMode, isTablePartitioned)
+ BulkInsertInternalPartitionerWithRowsFactory.get(writeConfig, isTablePartitioned)
}
} else {
// Sort modes are not yet supported when meta fields are disabled
@@ -886,7 +885,7 @@ object HoodieSparkSqlWriter {
}
private def mergeParamsAndGetHoodieConfig(optParams: Map[String, String],
- tableConfig: HoodieTableConfig, mode: SaveMode): (Map[String, String], HoodieConfig) = {
+ tableConfig: HoodieTableConfig, mode: SaveMode): (Map[String, String], HoodieConfig) = {
val translatedOptions = DataSourceWriteOptions.translateSqlOptions(optParams)
val mergedParams = mutable.Map.empty ++ HoodieWriterUtils.parametersWithWriteDefaults(translatedOptions)
if (!mergedParams.contains(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key)
[hudi] 04/08: [HUDI-5373] Different fileids are assigned to the same bucket (#7433)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d30b7d3857ce130f5ba4b8215c671f91298d77d0
Author: luokey <85...@qq.com>
AuthorDate: Mon Dec 12 21:40:35 2022 -0500
[HUDI-5373] Different fileids are assigned to the same bucket (#7433)
---
.../java/org/apache/hudi/sink/bucket/BucketStreamWriteFunction.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketStreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketStreamWriteFunction.java
index 1ccfe91dbc0..09a78de4c8d 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketStreamWriteFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketStreamWriteFunction.java
@@ -111,7 +111,7 @@ public class BucketStreamWriteFunction<I> extends StreamWriteFunction<I> {
bootstrapIndexIfNeed(partition);
Map<Integer, String> bucketToFileId = bucketIndex.computeIfAbsent(partition, p -> new HashMap<>());
final int bucketNum = BucketIdentifier.getBucketId(hoodieKey, indexKeyFields, this.bucketNum);
- final String bucketId = partition + bucketNum;
+ final String bucketId = partition + "/" + bucketNum;
if (incBucketIndex.contains(bucketId)) {
location = new HoodieRecordLocation("I", bucketToFileId.get(bucketNum));
[hudi] 02/08: [HUDI-5296] Allow disable schema on read after enabling (#7421)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 8a4c86b958254b8b06e5cd3269ef00f27908fb7b
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Mon Dec 12 11:39:18 2022 -0800
[HUDI-5296] Allow disable schema on read after enabling (#7421)
If someone has enabled schema on read by mistake and never really renamed or dropped a column. it should be feasible to disable schema on read. This patch fixes that. essentially both on read and write path, if "hoodie.schema.on.read.enable" config is not set, it will fallback to regular code path. It might fail or users might miss data if any they have performed any irrevocable changes like renames. But for rest, this should work.
---
.../scala/org/apache/hudi/HoodieBaseRelation.scala | 20 ++--
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 119 ++++++++++++++-------
.../org/apache/hudi/IncrementalRelation.scala | 6 +-
.../org/apache/spark/sql/hudi/TestSpark3DDL.scala | 54 +++++++++-
4 files changed, 146 insertions(+), 53 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
index 5e618f822d3..242d7eb8679 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
@@ -136,7 +136,7 @@ abstract class HoodieBaseRelation(val sqlContext: SQLContext,
*/
protected lazy val (tableAvroSchema: Schema, internalSchemaOpt: Option[InternalSchema]) = {
val schemaResolver = new TableSchemaResolver(metaClient)
- val internalSchemaOpt = if (!isSchemaEvolutionEnabled) {
+ val internalSchemaOpt = if (!isSchemaEvolutionEnabledOnRead(optParams, sparkSession)) {
None
} else {
Try {
@@ -635,15 +635,6 @@ abstract class HoodieBaseRelation(val sqlContext: SQLContext,
private def prunePartitionColumns(dataStructSchema: StructType): StructType =
StructType(dataStructSchema.filterNot(f => partitionColumns.contains(f.name)))
-
- private def isSchemaEvolutionEnabled = {
- // NOTE: Schema evolution could be configured both t/h optional parameters vehicle as well as
- // t/h Spark Session configuration (for ex, for Spark SQL)
- optParams.getOrElse(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key,
- DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
- sparkSession.conf.get(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key,
- DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean
- }
}
object HoodieBaseRelation extends SparkAdapterSupport {
@@ -745,4 +736,13 @@ object HoodieBaseRelation extends SparkAdapterSupport {
})
}
}
+
+ def isSchemaEvolutionEnabledOnRead(optParams: Map[String, String], sparkSession: SparkSession): Boolean = {
+ // NOTE: Schema evolution could be configured both t/h optional parameters vehicle as well as
+ // t/h Spark Session configuration (for ex, for Spark SQL)
+ optParams.getOrElse(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key,
+ DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
+ sparkSession.conf.get(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key,
+ DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean
+ }
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index 51cd30f7f23..003dbd35e11 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -21,6 +21,7 @@ import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
+import org.apache.hudi.AvroConversionUtils.{convertStructTypeToAvroSchema, getAvroRecordNameAndNamespace}
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.HoodieConversionUtils.{toProperties, toScalaOption}
import org.apache.hudi.HoodieWriterUtils._
@@ -55,6 +56,8 @@ import org.apache.log4j.LogManager
import org.apache.spark.api.java.JavaSparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.internal.StaticSQLConf
import org.apache.spark.sql.types.StructType
import org.apache.spark.{SPARK_VERSION, SparkContext}
@@ -84,6 +87,7 @@ object HoodieSparkSqlWriter {
assert(optParams.get("path").exists(!StringUtils.isNullOrEmpty(_)), "'path' must be set")
val path = optParams("path")
val basePath = new Path(path)
+ val spark = sqlContext.sparkSession
val sparkContext = sqlContext.sparkContext
val fs = basePath.getFileSystem(sparkContext.hadoopConfiguration)
tableExists = fs.exists(new Path(basePath, HoodieTableMetaClient.METAFOLDER_NAME))
@@ -98,6 +102,7 @@ object HoodieSparkSqlWriter {
val databaseName = hoodieConfig.getStringOrDefault(HoodieTableConfig.DATABASE_NAME, "")
val tblName = hoodieConfig.getStringOrThrow(HoodieWriteConfig.TBL_NAME,
s"'${HoodieWriteConfig.TBL_NAME.key}' must be set.").trim
+ val tableIdentifier = TableIdentifier(tblName, if (databaseName.isEmpty) None else Some(databaseName))
assert(!StringUtils.isNullOrEmpty(hoodieConfig.getString(HoodieWriteConfig.TBL_NAME)),
s"'${HoodieWriteConfig.TBL_NAME.key}' must be set.")
@@ -139,15 +144,18 @@ object HoodieSparkSqlWriter {
// Handle various save modes
handleSaveModes(sqlContext.sparkSession, mode, basePath, tableConfig, tblName, operation, fs)
val partitionColumns = SparkKeyGenUtils.getPartitionColumns(keyGenerator, toProperties(parameters))
- // Create the table if not present
- if (!tableExists) {
+ val tableMetaClient = if (tableExists) {
+ HoodieTableMetaClient.builder
+ .setConf(sparkContext.hadoopConfiguration)
+ .setBasePath(path)
+ .build()
+ } else {
val baseFileFormat = hoodieConfig.getStringOrDefault(HoodieTableConfig.BASE_FILE_FORMAT)
val archiveLogFolder = hoodieConfig.getStringOrDefault(HoodieTableConfig.ARCHIVELOG_FOLDER)
val recordKeyFields = hoodieConfig.getString(DataSourceWriteOptions.RECORDKEY_FIELD)
val populateMetaFields = hoodieConfig.getBooleanOrDefault(HoodieTableConfig.POPULATE_META_FIELDS)
val useBaseFormatMetaFile = hoodieConfig.getBooleanOrDefault(HoodieTableConfig.PARTITION_METAFILE_USE_BASE_FORMAT);
-
- val tableMetaClient = HoodieTableMetaClient.withPropertyBuilder()
+ HoodieTableMetaClient.withPropertyBuilder()
.setTableType(tableType)
.setDatabaseName(databaseName)
.setTableName(tblName)
@@ -169,8 +177,8 @@ object HoodieSparkSqlWriter {
.setShouldDropPartitionColumns(hoodieConfig.getBooleanOrDefault(HoodieTableConfig.DROP_PARTITION_COLUMNS))
.setCommitTimezone(HoodieTimelineTimeZone.valueOf(hoodieConfig.getStringOrDefault(HoodieTableConfig.TIMELINE_TIMEZONE)))
.initTable(sparkContext.hadoopConfiguration, path)
- tableConfig = tableMetaClient.getTableConfig
- }
+ }
+ tableConfig = tableMetaClient.getTableConfig
val commitActionType = CommitUtils.getCommitActionType(operation, tableConfig.getTableType)
val dropPartitionColumns = hoodieConfig.getBoolean(DataSourceWriteOptions.DROP_PARTITION_COLUMNS)
@@ -198,7 +206,7 @@ object HoodieSparkSqlWriter {
}
// Create a HoodieWriteClient & issue the delete.
- val internalSchemaOpt = getLatestTableInternalSchema(fs, basePath, sparkContext)
+ val internalSchemaOpt = getLatestTableInternalSchema(hoodieConfig, tableMetaClient)
val client = hoodieWriteClient.getOrElse(DataSourceUtils.createHoodieClient(jsc,
null, path, tblName,
mapAsJavaMap(addSchemaEvolutionParameters(parameters, internalSchemaOpt) - HoodieWriteConfig.AUTO_COMMIT_ENABLE.key)))
@@ -250,10 +258,10 @@ object HoodieSparkSqlWriter {
// TODO(HUDI-4472) revisit and simplify schema handling
val sourceSchema = AvroConversionUtils.convertStructTypeToAvroSchema(df.schema, structName, nameSpace)
- val latestTableSchema = getLatestTableSchema(fs, basePath, sparkContext).getOrElse(sourceSchema)
+ val latestTableSchema = getLatestTableSchema(spark, tableIdentifier, tableMetaClient).getOrElse(sourceSchema)
val schemaEvolutionEnabled = parameters.getOrDefault(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key(), "false").toBoolean
- var internalSchemaOpt = getLatestTableInternalSchema(fs, basePath, sparkContext)
+ var internalSchemaOpt = getLatestTableInternalSchema(hoodieConfig, tableMetaClient)
val writerSchema: Schema =
if (reconcileSchema) {
@@ -286,6 +294,34 @@ object HoodieSparkSqlWriter {
validateSchemaForHoodieIsDeleted(writerSchema)
sparkContext.getConf.registerAvroSchemas(writerSchema)
log.info(s"Registered avro schema : ${writerSchema.toString(true)}")
+/*=======
+ case _ =>
+ // Here all other (than DELETE, DELETE_PARTITION) write operations are handled
+ //
+ // Convert to RDD[HoodieRecord]
+ val avroRecords: RDD[GenericRecord] = HoodieSparkUtils.createRdd(df, avroRecordName, avroRecordNamespace,
+ Some(writerSchema))
+
+ // Check whether partition columns should be persisted w/in the data-files, or should
+ // be instead omitted from them and simply encoded into the partition path (which is Spark's
+ // behavior by default)
+ // TODO move partition columns handling down into the handlers
+ val shouldDropPartitionColumns = hoodieConfig.getBoolean(DataSourceWriteOptions.DROP_PARTITION_COLUMNS)
+ val dataFileSchema = if (shouldDropPartitionColumns) {
+ val truncatedSchema = generateSchemaWithoutPartitionColumns(partitionColumns, writerSchema)
+ // NOTE: We have to register this schema w/ Kryo to make sure it's able to apply an optimization
+ // allowing it to avoid the need to ser/de the whole schema along _every_ Avro record
+ registerAvroSchemasWithKryo(sparkContext, truncatedSchema)
+ truncatedSchema
+ } else {
+ writerSchema
+ }
+
+ // NOTE: Avro's [[Schema]] can't be effectively serialized by JVM native serialization framework
+ // (due to containing cyclic refs), therefore we have to convert it to string before
+ // passing onto the Executor
+ val dataFileSchemaStr = dataFileSchema.toString
+>>>>>>> aacfe6de80 ([HUDI-5296] Allow disable schema on read after enabling (#7421))*/
// Convert to RDD[HoodieRecord]
val genericRecords: RDD[GenericRecord] = HoodieSparkUtils.createRdd(df, structName, nameSpace, reconcileSchema,
@@ -407,47 +443,48 @@ object HoodieSparkSqlWriter {
}
/**
- * get latest internalSchema from table
- *
- * @param fs instance of FileSystem.
- * @param basePath base path.
- * @param sparkContext instance of spark context.
- * @param schema incoming record's schema.
- * @return Pair of(boolean, table schema), where first entry will be true only if schema conversion is required.
- */
- def getLatestTableInternalSchema(fs: FileSystem, basePath: Path, sparkContext: SparkContext): Option[InternalSchema] = {
- try {
- if (FSUtils.isTableExists(basePath.toString, fs)) {
- val tableMetaClient = HoodieTableMetaClient.builder.setConf(sparkContext.hadoopConfiguration).setBasePath(basePath.toString).build()
+ * get latest internalSchema from table
+ *
+ * @param config instance of {@link HoodieConfig}
+ * @param tableMetaClient instance of HoodieTableMetaClient
+ * @return Pair of(boolean, table schema), where first entry will be true only if schema conversion is required.
+ */
+ def getLatestTableInternalSchema(config: HoodieConfig,
+ tableMetaClient: HoodieTableMetaClient): Option[InternalSchema] = {
+ if (!config.getBooleanOrDefault(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED)) {
+ Option.empty[InternalSchema]
+ } else {
+ try {
val tableSchemaResolver = new TableSchemaResolver(tableMetaClient)
val internalSchemaOpt = tableSchemaResolver.getTableInternalSchemaFromCommitMetadata
if (internalSchemaOpt.isPresent) Some(internalSchemaOpt.get()) else None
- } else {
- None
+ } catch {
+ case _: Exception => None
}
- } catch {
- case _: Exception => None
}
}
- /**
- * Checks if schema needs upgrade (if incoming record's write schema is old while table schema got evolved).
- *
- * @param fs instance of FileSystem.
- * @param basePath base path.
- * @param sparkContext instance of spark context.
- * @param schema incoming record's schema.
- * @return Pair of(boolean, table schema), where first entry will be true only if schema conversion is required.
- */
- def getLatestTableSchema(fs: FileSystem, basePath: Path, sparkContext: SparkContext): Option[Schema] = {
- if (FSUtils.isTableExists(basePath.toString, fs)) {
- val tableMetaClient = HoodieTableMetaClient.builder
- .setConf(sparkContext.hadoopConfiguration)
- .setBasePath(basePath.toString)
- .build()
- val tableSchemaResolver = new TableSchemaResolver(tableMetaClient)
+ private def registerAvroSchemasWithKryo(sparkContext: SparkContext, targetAvroSchemas: Schema*): Unit = {
+ sparkContext.getConf.registerAvroSchemas(targetAvroSchemas: _*)
+ }
+ private def getLatestTableSchema(spark: SparkSession,
+ tableId: TableIdentifier,
+ tableMetaClient: HoodieTableMetaClient): Option[Schema] = {
+ val tableSchemaResolver = new TableSchemaResolver(tableMetaClient)
+ val latestTableSchemaFromCommitMetadata =
toScalaOption(tableSchemaResolver.getTableAvroSchemaFromLatestCommit(false))
+ latestTableSchemaFromCommitMetadata.orElse {
+ getCatalogTable(spark, tableId).map { catalogTable =>
+ val (structName, namespace) = getAvroRecordNameAndNamespace(tableId.table)
+ convertStructTypeToAvroSchema(catalogTable.schema, structName, namespace)
+ }
+ }
+ }
+
+ private def getCatalogTable(spark: SparkSession, tableId: TableIdentifier): Option[CatalogTable] = {
+ if (spark.sessionState.catalog.tableExists(tableId)) {
+ Some(spark.sessionState.catalog.getTableMetadata(tableId))
} else {
None
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala
index 4c763b054ad..80ee3dde5b0 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelation.scala
@@ -19,6 +19,7 @@ package org.apache.hudi
import org.apache.avro.Schema
import org.apache.hadoop.fs.{GlobPattern, Path}
+import org.apache.hudi.HoodieBaseRelation.isSchemaEvolutionEnabledOnRead
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.fs.FSUtils
@@ -47,6 +48,7 @@ import scala.collection.mutable
* Relation, that implements the Hoodie incremental view.
*
* Implemented for Copy_on_write storage.
+ * TODO: rebase w/ HoodieBaseRelation HUDI-5362
*
*/
class IncrementalRelation(val sqlContext: SQLContext,
@@ -90,7 +92,9 @@ class IncrementalRelation(val sqlContext: SQLContext,
val (usedSchema, internalSchema) = {
log.info("Inferring schema..")
val schemaResolver = new TableSchemaResolver(metaClient)
- val iSchema = if (useEndInstantSchema && !commitsToReturn.isEmpty) {
+ val iSchema : InternalSchema = if (!isSchemaEvolutionEnabledOnRead(optParams, sqlContext.sparkSession)) {
+ InternalSchema.getEmptyInternalSchema
+ } else if (useEndInstantSchema && !commitsToReturn.isEmpty) {
InternalSchemaCache.searchSchemaAndCache(commitsToReturn.last.getTimestamp.toLong, metaClient, hoodieTable.getConfig.getInternalSchemaCacheEnable)
} else {
schemaResolver.getTableInternalSchemaFromCommitMetadata.orElse(null)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestSpark3DDL.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestSpark3DDL.scala
index 9d955cb8310..056bca37fce 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestSpark3DDL.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestSpark3DDL.scala
@@ -22,7 +22,7 @@ import org.apache.hudi.common.model.HoodieRecord
import org.apache.hudi.common.testutils.HoodieTestDataGenerator
import org.apache.hudi.common.testutils.RawTripTestPayload
import org.apache.hudi.config.HoodieWriteConfig
-import org.apache.hudi.{DataSourceWriteOptions, HoodieSparkUtils}
+import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions, HoodieSparkUtils}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.functions.{arrays_zip, col}
import org.apache.spark.sql.{Row, SaveMode, SparkSession}
@@ -171,6 +171,58 @@ class TestSpark3DDL extends HoodieSparkSqlTestBase {
}
}
+ test("Test Enable and Disable Schema on read") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ if (HoodieSparkUtils.gteqSpark3_1) {
+ spark.sql("set hoodie.schema.on.read.enable=true")
+ // Create table
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | preCombineField = 'ts'
+ | )
+ """.stripMargin)
+
+ // Insert data to the new table.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ checkAnswer(s"select id, name, price, ts from $tableName")(
+ Seq(1, "a1", 10.0, 1000)
+ )
+
+ // add column
+ spark.sql(s"alter table $tableName add columns(new_col string)")
+ val catalogTable = spark.sessionState.catalog.getTableMetadata(new TableIdentifier(tableName))
+ assertResult(Seq("id", "name", "price", "ts", "new_col")) {
+ HoodieSqlCommonUtils.removeMetaFields(catalogTable.schema).fields.map(_.name)
+ }
+ checkAnswer(s"select id, name, price, ts, new_col from $tableName")(
+ Seq(1, "a1", 10.0, 1000, null)
+ )
+ // disable schema on read.
+ spark.sql("set hoodie.schema.on.read.enable=false")
+ spark.sql(s"refresh table $tableName")
+ // Insert data to the new table.
+ spark.sql(s"insert into $tableName values(2, 'a2', 12, 2000, 'e0')")
+ // write should succeed. and subsequent read should succeed as well.
+ checkAnswer(s"select id, name, price, ts, new_col from $tableName")(
+ Seq(1, "a1", 10.0, 1000, null),
+ Seq(2, "a2", 12.0, 2000, "e0")
+ )
+ }
+ }
+ }
+
test("Test Partition Table alter ") {
withTempDir { tmp =>
Seq("cow", "mor").foreach { tableType =>
[hudi] 07/08: [HUDI-4113] Fix cannot parse schema when use spark delete sql (#5610)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a56c00e767f5bd64beb533263ac4b9dee3c305c1
Author: KnightChess <98...@qq.com>
AuthorDate: Tue Dec 13 23:02:09 2022 +0800
[HUDI-4113] Fix cannot parse <null> schema when use spark delete sql (#5610)
- Alter table drop partition will not add schema to instant.
If using delete sql, will get latest instant to get schema,
which is "". This PR fixes the parsing of null or empty schema.
Co-authored-by: Sagar Sumit <sa...@gmail.com>
---
.../apache/hudi/client/BaseHoodieWriteClient.java | 4 +-
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 5 +-
.../sql/hudi/TestAlterTableDropPartition.scala | 62 +++++++++++++++++++++-
3 files changed, 68 insertions(+), 3 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index 011235e4369..56ad47c9453 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -59,6 +59,7 @@ import org.apache.hudi.common.util.CleanerUtils;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieArchivalConfig;
@@ -1513,7 +1514,8 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
if (lastInstant.isPresent()) {
HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(
activeTimeline.getInstantDetails(lastInstant.get()).get(), HoodieCommitMetadata.class);
- if (commitMetadata.getExtraMetadata().containsKey(SCHEMA_KEY)) {
+ String extraSchema = commitMetadata.getExtraMetadata().get(SCHEMA_KEY);
+ if (!StringUtils.isNullOrEmpty(extraSchema)) {
config.setSchema(commitMetadata.getExtraMetadata().get(SCHEMA_KEY));
} else {
throw new HoodieIOException("Latest commit does not have any schema in commit metadata");
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index d8f9ab8435b..b358ae1d2d0 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -240,8 +240,11 @@ object HoodieSparkSqlWriter {
}
// Create a HoodieWriteClient & issue the delete.
+ val tableMetaClient = HoodieTableMetaClient.builder
+ .setConf(sparkContext.hadoopConfiguration).setBasePath(basePath.toString).build()
+ val schemaStr = new TableSchemaResolver(tableMetaClient).getTableAvroSchema.toString
val client = hoodieWriteClient.getOrElse(DataSourceUtils.createHoodieClient(jsc,
- null, path, tblName,
+ schemaStr, path, tblName,
mapAsJavaMap(parameters - HoodieWriteConfig.AUTO_COMMIT_ENABLE.key)))
.asInstanceOf[SparkRDDWriteClient[HoodieRecordPayload[Nothing]]]
// Issue delete partitions
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestAlterTableDropPartition.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestAlterTableDropPartition.scala
index e063f67d8c0..15b14ec77f5 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestAlterTableDropPartition.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestAlterTableDropPartition.scala
@@ -19,10 +19,13 @@ package org.apache.spark.sql.hudi
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.HoodieSparkUtils
-import org.apache.hudi.common.util.PartitionPathEncodeUtils
+import org.apache.hudi.common.model.HoodieCommitMetadata
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.util.{PartitionPathEncodeUtils, StringUtils}
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.keygen.{ComplexKeyGenerator, SimpleKeyGenerator}
import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.Assertions
class TestAlterTableDropPartition extends HoodieSparkSqlTestBase {
@@ -336,4 +339,61 @@ class TestAlterTableDropPartition extends HoodieSparkSqlTestBase {
}
}
}
+
+ test("check instance schema") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ val tableName = generateTableName
+ // create table
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long,
+ | dt string
+ |) using hudi
+ | location '${tmp.getCanonicalPath}/$tableName'
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | preCombineField = 'ts'
+ | )
+ | partitioned by (dt)
+ """.stripMargin)
+ // insert data to table
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000, '01'), " +
+ s"(2, 'a2', 10, 1000, '02'), (3, 'a3', 10, 1000, '03')")
+ checkAnswer(s"select id, name, price, ts, dt from $tableName")(
+ Seq(1, "a1", 10.0, 1000, "01"),
+ Seq(2, "a2", 10.0, 1000, "02"),
+ Seq(3, "a3", 10.0, 1000, "03")
+ )
+
+ // drop partition
+ spark.sql(s"alter table $tableName drop partition (dt = '01')")
+ checkAnswer(s"select id, name, price, ts, dt from $tableName")(
+ Seq(2, "a2", 10.0, 1000, "02"),
+ Seq(3, "a3", 10.0, 1000, "03")
+ )
+
+ // check schema
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val metaClient = HoodieTableMetaClient.builder().setBasePath(s"${tmp.getCanonicalPath}/$tableName")
+ .setConf(hadoopConf).build()
+ val lastInstant = metaClient.getActiveTimeline.lastInstant()
+ val commitMetadata = HoodieCommitMetadata.fromBytes(metaClient.getActiveTimeline.getInstantDetails(
+ lastInstant.get()).get(), classOf[HoodieCommitMetadata])
+ val schemaStr = commitMetadata.getExtraMetadata.get(HoodieCommitMetadata.SCHEMA_KEY)
+ Assertions.assertFalse(StringUtils.isNullOrEmpty(schemaStr))
+
+ // delete
+ spark.sql(s"delete from $tableName where dt = '02'")
+ checkAnswer(s"select id, name, price, ts, dt from $tableName")(
+ Seq(3, "a3", 10, 1000, "03")
+ )
+ }
+ }
+ }
}
[hudi] 05/08: 【HUDI-4917】Optimized the way to get HoodieBaseFile of loadColumnRangesFromFiles of Bloom Index (#6793)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 20956a21ac57366acdc75c81b0381a0e4ab10848
Author: LiChuang <64...@users.noreply.github.com>
AuthorDate: Tue Dec 13 12:37:38 2022 +0800
【HUDI-4917】Optimized the way to get HoodieBaseFile of loadColumnRangesFromFiles of Bloom Index (#6793)
---
.../java/org/apache/hudi/index/bloom/HoodieBloomIndex.java | 13 +++++++------
.../main/java/org/apache/hudi/io/HoodieRangeInfoHandle.java | 8 ++++++++
.../src/main/java/org/apache/hudi/io/HoodieReadHandle.java | 5 +++++
3 files changed, 20 insertions(+), 6 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
index 1417e40a9f5..57d9def9b42 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
@@ -26,6 +26,7 @@ import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
@@ -161,19 +162,19 @@ public class HoodieBloomIndex extends HoodieIndex<Object, Object> {
List<Pair<String, BloomIndexFileInfo>> loadColumnRangesFromFiles(
List<String> partitions, final HoodieEngineContext context, final HoodieTable hoodieTable) {
// Obtain the latest data files from all the partitions.
- List<Pair<String, String>> partitionPathFileIDList = getLatestBaseFilesForAllPartitions(partitions, context, hoodieTable).stream()
- .map(pair -> Pair.of(pair.getKey(), pair.getValue().getFileId()))
+ List<Pair<String, Pair<String, HoodieBaseFile>>> partitionPathFileIDList = getLatestBaseFilesForAllPartitions(partitions, context, hoodieTable).stream()
+ .map(pair -> Pair.of(pair.getKey(), Pair.of(pair.getValue().getFileId(), pair.getValue())))
.collect(toList());
context.setJobStatus(this.getClass().getName(), "Obtain key ranges for file slices (range pruning=on): " + config.getTableName());
return context.map(partitionPathFileIDList, pf -> {
try {
- HoodieRangeInfoHandle rangeInfoHandle = new HoodieRangeInfoHandle(config, hoodieTable, pf);
- String[] minMaxKeys = rangeInfoHandle.getMinMaxKeys();
- return Pair.of(pf.getKey(), new BloomIndexFileInfo(pf.getValue(), minMaxKeys[0], minMaxKeys[1]));
+ HoodieRangeInfoHandle rangeInfoHandle = new HoodieRangeInfoHandle(config, hoodieTable, Pair.of(pf.getKey(), pf.getValue().getKey()));
+ String[] minMaxKeys = rangeInfoHandle.getMinMaxKeys(pf.getValue().getValue());
+ return Pair.of(pf.getKey(), new BloomIndexFileInfo(pf.getValue().getKey(), minMaxKeys[0], minMaxKeys[1]));
} catch (MetadataNotFoundException me) {
LOG.warn("Unable to find range metadata in file :" + pf);
- return Pair.of(pf.getKey(), new BloomIndexFileInfo(pf.getValue()));
+ return Pair.of(pf.getKey(), new BloomIndexFileInfo(pf.getValue().getKey()));
}
}, Math.max(partitionPathFileIDList.size(), 1));
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieRangeInfoHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieRangeInfoHandle.java
index abe4a9befef..44407335081 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieRangeInfoHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieRangeInfoHandle.java
@@ -18,6 +18,7 @@
package org.apache.hudi.io;
+import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -41,4 +42,11 @@ public class HoodieRangeInfoHandle<T extends HoodieRecordPayload, I, K, O> exten
return reader.readMinMaxRecordKeys();
}
}
+
+ public String[] getMinMaxKeys(HoodieBaseFile baseFile) throws IOException {
+ try (HoodieFileReader reader = createNewFileReader(baseFile)) {
+ return reader.readMinMaxRecordKeys();
+ }
+ }
+
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java
index fee75b22dec..223241bc373 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieReadHandle.java
@@ -66,4 +66,9 @@ public abstract class HoodieReadHandle<T extends HoodieRecordPayload, I, K, O> e
return HoodieFileReaderFactory.getFileReader(hoodieTable.getHadoopConf(),
new Path(getLatestDataFile().getPath()));
}
+
+ protected HoodieFileReader createNewFileReader(HoodieBaseFile hoodieBaseFile) throws IOException {
+ return HoodieFileReaderFactory.getFileReader(hoodieTable.getHadoopConf(),
+ new Path(hoodieBaseFile.getPath()));
+ }
}
[hudi] 03/08: [HUDI-5348] Cache file slices in HoodieBackedTableMetadata (#7436)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 16a89240b2562e15ce70febee0229425824753cd
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Mon Dec 12 17:00:10 2022 -0800
[HUDI-5348] Cache file slices in HoodieBackedTableMetadata (#7436)
---
.../org/apache/hudi/metadata/HoodieBackedTableMetadata.java | 13 +++++++++++--
.../org/apache/hudi/metadata/HoodieTableMetadataUtil.java | 10 ++++++----
.../java/org/apache/hudi/utilities/TestHoodieIndexer.java | 7 +++++--
3 files changed, 22 insertions(+), 8 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
index 11b25b4aad0..4d36bd58dc8 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
@@ -40,6 +40,7 @@ import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.ClosableIterator;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
@@ -78,6 +79,7 @@ import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_BLOOM_FILTERS;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_FILES;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getFileSystemView;
/**
* Table metadata provided by an internal DFS backed Hudi metadata table.
@@ -92,6 +94,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// Metadata table's timeline and metaclient
private HoodieTableMetaClient metadataMetaClient;
private HoodieTableConfig metadataTableConfig;
+ private HoodieTableFileSystemView metadataFileSystemView;
// should we reuse the open file handles, across calls
private final boolean reuse;
@@ -120,6 +123,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
} else if (this.metadataMetaClient == null) {
try {
this.metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf.get()).setBasePath(metadataBasePath).build();
+ this.metadataFileSystemView = getFileSystemView(metadataMetaClient);
this.metadataTableConfig = metadataMetaClient.getTableConfig();
this.isBloomFilterIndexEnabled = metadataConfig.isBloomFilterIndexEnabled();
this.isColumnStatsIndexEnabled = metadataConfig.isColumnStatsIndexEnabled();
@@ -127,11 +131,13 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
LOG.warn("Metadata table was not found at path " + metadataBasePath);
this.isMetadataTableEnabled = false;
this.metadataMetaClient = null;
+ this.metadataFileSystemView = null;
this.metadataTableConfig = null;
} catch (Exception e) {
LOG.error("Failed to initialize metadata table at path " + metadataBasePath, e);
this.isMetadataTableEnabled = false;
this.metadataMetaClient = null;
+ this.metadataFileSystemView = null;
this.metadataTableConfig = null;
}
}
@@ -155,7 +161,8 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// to scan all file-groups for all key-prefixes as each of these might contain some
// records matching the key-prefix
List<FileSlice> partitionFileSlices =
- HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, partitionName);
+ HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(
+ metadataMetaClient, metadataFileSystemView, partitionName);
return (shouldLoadInMemory ? HoodieListData.lazy(partitionFileSlices) : engineContext.parallelize(partitionFileSlices))
.flatMap((SerializableFunction<FileSlice, Iterator<HoodieRecord<HoodieMetadataPayload>>>) fileSlice -> {
@@ -372,7 +379,8 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
private Map<Pair<String, FileSlice>, List<String>> getPartitionFileSliceToKeysMapping(final String partitionName, final List<String> keys) {
// Metadata is in sync till the latest completed instant on the dataset
List<FileSlice> latestFileSlices =
- HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, partitionName);
+ HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(
+ metadataMetaClient, metadataFileSystemView, partitionName);
Map<Pair<String, FileSlice>, List<String>> partitionFileSliceToKeysMap = new HashMap<>();
for (String key : keys) {
@@ -638,6 +646,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
dataMetaClient.reloadActiveTimeline();
if (metadataMetaClient != null) {
metadataMetaClient.reloadActiveTimeline();
+ metadataFileSystemView = getFileSystemView(metadataMetaClient);
}
// the cached reader has max instant time restriction, they should be cleared
// because the metadata timeline may have changed.
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 8e7e2a41588..626bee2e38e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -987,13 +987,15 @@ public class HoodieTableMetadataUtil {
* just before the compaction instant time. The list of file slices returned is
* sorted in the correct order of file group name.
*
- * @param metaClient - Instance of {@link HoodieTableMetaClient}.
- * @param partition - The name of the partition whose file groups are to be loaded.
+ * @param metaClient Instance of {@link HoodieTableMetaClient}.
+ * @param fsView Metadata table filesystem view.
+ * @param partition The name of the partition whose file groups are to be loaded.
* @return List of latest file slices for all file groups in a given partition.
*/
- public static List<FileSlice> getPartitionLatestMergedFileSlices(HoodieTableMetaClient metaClient, String partition) {
+ public static List<FileSlice> getPartitionLatestMergedFileSlices(
+ HoodieTableMetaClient metaClient, HoodieTableFileSystemView fsView, String partition) {
LOG.info("Loading latest merged file slices for metadata table partition " + partition);
- return getPartitionFileSlices(metaClient, Option.empty(), partition, true);
+ return getPartitionFileSlices(metaClient, Option.of(fsView), partition, true);
}
/**
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 f5a0fadc87f..ac7b86f4cfa 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
@@ -56,6 +56,7 @@ import java.util.stream.Stream;
import static org.apache.hudi.common.table.HoodieTableMetaClient.reload;
import static org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getFileSystemView;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.metadataPartitionExists;
import static org.apache.hudi.metadata.MetadataPartitionType.BLOOM_FILTERS;
import static org.apache.hudi.metadata.MetadataPartitionType.COLUMN_STATS;
@@ -175,7 +176,8 @@ public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implemen
HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(metaClient.getHadoopConf()).setBasePath(metaClient.getMetaPath() + "/metadata").build();
List<FileSlice> partitionFileSlices =
- HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, COLUMN_STATS.getPartitionPath());
+ HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(
+ metadataMetaClient, getFileSystemView(metadataMetaClient), COLUMN_STATS.getPartitionPath());
assertEquals(partitionFileSlices.size(), colStatsFileGroupCount);
}
@@ -220,7 +222,8 @@ public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implemen
HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(metaClient.getHadoopConf()).setBasePath(metaClient.getMetaPath() + "/metadata").build();
List<FileSlice> partitionFileSlices =
- HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, COLUMN_STATS.getPartitionPath());
+ HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(
+ metadataMetaClient, getFileSystemView(metadataMetaClient), COLUMN_STATS.getPartitionPath());
assertEquals(partitionFileSlices.size(), HoodieMetadataConfig.METADATA_INDEX_COLUMN_STATS_FILE_GROUP_COUNT.defaultValue());
}
[hudi] 01/08: [HUDI-5078] Fixing isTableService for replace commits (#7037)
Posted by si...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.12.2-shadow
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 38e5a35e76e9def6acfa1db5fa2858ecedccc867
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Mon Dec 12 07:28:22 2022 -0800
[HUDI-5078] Fixing isTableService for replace commits (#7037)
---
.../apache/hudi/client/BaseHoodieWriteClient.java | 2 +-
.../java/org/apache/hudi/table/HoodieTable.java | 19 ++++++++++++++++--
.../hudi/table/action/BaseActionExecutor.java | 2 +-
.../apache/hudi/client/HoodieFlinkWriteClient.java | 2 +-
.../hudi/table/HoodieFlinkCopyOnWriteTable.java | 6 ------
.../hudi/table/HoodieFlinkMergeOnReadTable.java | 6 ------
.../hudi/table/HoodieJavaCopyOnWriteTable.java | 6 ------
.../hudi/table/HoodieJavaMergeOnReadTable.java | 6 ------
.../apache/hudi/client/SparkRDDWriteClient.java | 2 +-
.../hudi/table/HoodieSparkCopyOnWriteTable.java | 5 -----
.../hudi/table/HoodieSparkMergeOnReadTable.java | 5 -----
.../apache/hudi/io/TestHoodieTimelineArchiver.java | 23 ++++++++++++++++++++++
.../hudi/common/table/timeline/HoodieInstant.java | 2 +-
.../table/timeline/TestHoodieActiveTimeline.java | 2 +-
14 files changed, 46 insertions(+), 42 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index 609f85e27fe..011235e4369 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -348,7 +348,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
protected void writeTableMetadata(HoodieTable table, String instantTime, String actionType, HoodieCommitMetadata metadata) {
context.setJobStatus(this.getClass().getSimpleName(), "Committing to metadata table: " + config.getTableName());
table.getMetadataWriter(instantTime).ifPresent(w -> ((HoodieTableMetadataWriter) w).update(metadata, instantTime,
- table.isTableServiceAction(actionType)));
+ table.isTableServiceAction(actionType, instantTime)));
}
/**
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
index 77b24a38953..7a9211a7c57 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
@@ -44,6 +44,7 @@ import org.apache.hudi.common.fs.OptimisticConsistencyGuard;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -58,6 +59,7 @@ import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.common.table.view.TableFileSystemView;
import org.apache.hudi.common.table.view.TableFileSystemView.BaseFileOnlyView;
import org.apache.hudi.common.table.view.TableFileSystemView.SliceView;
+import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Functions;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
@@ -823,10 +825,23 @@ public abstract class HoodieTable<T extends HoodieRecordPayload, I, K, O> implem
/**
* Check if action type is a table service.
- * @param actionType action type of interest.
+ * @param actionType action type of the instant
+ * @param instantTime instant time of the instant.
* @return true if action represents a table service. false otherwise.
*/
- public abstract boolean isTableServiceAction(String actionType);
+ public boolean isTableServiceAction(String actionType, String instantTime) {
+ if (actionType.equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
+ Option<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlan = ClusteringUtils.getClusteringPlan(metaClient, new HoodieInstant(HoodieInstant.State.NIL, actionType, instantTime));
+ // only clustering is table service with replace commit action
+ return instantPlan.isPresent();
+ } else {
+ if (this.metaClient.getTableType() == HoodieTableType.COPY_ON_WRITE) {
+ return !actionType.equals(HoodieTimeline.COMMIT_ACTION);
+ } else {
+ return !actionType.equals(HoodieTimeline.DELTA_COMMIT_ACTION);
+ }
+ }
+ }
/**
* Get Table metadata writer.
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseActionExecutor.java
index f893b4ccd5c..d2b2cef2f60 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseActionExecutor.java
@@ -58,7 +58,7 @@ public abstract class BaseActionExecutor<T extends HoodieRecordPayload, I, K, O,
*/
protected final void writeTableMetadata(HoodieCommitMetadata metadata, String actionType) {
table.getMetadataWriter(instantTime).ifPresent(w -> w.update(
- metadata, instantTime, table.isTableServiceAction(actionType)));
+ metadata, instantTime, table.isTableServiceAction(actionType, instantTime)));
}
/**
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
index 33148cddb11..b4003712c56 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -289,7 +289,7 @@ public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
// the schema expects to be immutable for SQL jobs but may be not for non-SQL
// jobs.
this.metadataWriter.initTableMetadata();
- this.metadataWriter.update(metadata, instantTime, getHoodieTable().isTableServiceAction(actionType));
+ this.metadataWriter.update(metadata, instantTime, getHoodieTable().isTableServiceAction(actionType, instantTime));
}
/**
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
index 2c8a3c4e49c..543751a0410 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
@@ -38,7 +38,6 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieIOException;
@@ -93,11 +92,6 @@ public class HoodieFlinkCopyOnWriteTable<T extends HoodieRecordPayload>
super(config, context, metaClient);
}
- @Override
- public boolean isTableServiceAction(String actionType) {
- return !actionType.equals(HoodieTimeline.COMMIT_ACTION);
- }
-
/**
* Upsert a batch of new records into Hoodie table at the supplied instantTime.
*
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkMergeOnReadTable.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkMergeOnReadTable.java
index aa8adde7353..9b7d3447177 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkMergeOnReadTable.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkMergeOnReadTable.java
@@ -27,7 +27,6 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -58,11 +57,6 @@ public class HoodieFlinkMergeOnReadTable<T extends HoodieRecordPayload>
super(config, context, metaClient);
}
- @Override
- public boolean isTableServiceAction(String actionType) {
- return !actionType.equals(HoodieTimeline.DELTA_COMMIT_ACTION);
- }
-
@Override
public HoodieWriteMetadata<List<WriteStatus>> upsert(
HoodieEngineContext context,
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java
index 88921334980..342c018e5a2 100644
--- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java
+++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java
@@ -38,7 +38,6 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieNotSupportedException;
@@ -90,11 +89,6 @@ public class HoodieJavaCopyOnWriteTable<T extends HoodieRecordPayload>
super(config, context, metaClient);
}
- @Override
- public boolean isTableServiceAction(String actionType) {
- return !actionType.equals(HoodieTimeline.COMMIT_ACTION);
- }
-
@Override
public HoodieWriteMetadata<List<WriteStatus>> upsert(HoodieEngineContext context,
String instantTime,
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaMergeOnReadTable.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaMergeOnReadTable.java
index 32d30f704ec..5af29502a95 100644
--- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaMergeOnReadTable.java
+++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaMergeOnReadTable.java
@@ -25,7 +25,6 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.action.HoodieWriteMetadata;
@@ -43,11 +42,6 @@ public class HoodieJavaMergeOnReadTable<T extends HoodieRecordPayload> extends H
super(config, context, metaClient);
}
- @Override
- public boolean isTableServiceAction(String actionType) {
- return !actionType.equals(HoodieTimeline.DELTA_COMMIT_ACTION);
- }
-
@Override
public HoodieWriteMetadata<List<WriteStatus>> upsertPrepped(HoodieEngineContext context,
String instantTime,
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
index ea7a660c379..fa568a7a6fe 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -425,7 +425,7 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
private void updateTableMetadata(HoodieTable table, HoodieCommitMetadata commitMetadata,
HoodieInstant hoodieInstant) {
- boolean isTableServiceAction = table.isTableServiceAction(hoodieInstant.getAction());
+ boolean isTableServiceAction = table.isTableServiceAction(hoodieInstant.getAction(), hoodieInstant.getTimestamp());
// Do not do any conflict resolution here as we do with regular writes. We take the lock here to ensure all writes to metadata table happens within a
// single lock (single writer). Because more than one write to metadata table will result in conflicts since all of them updates the same partition.
table.getMetadataWriter(hoodieInstant.getTimestamp())
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
index a88ca65c35a..743aff51a12 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
@@ -103,11 +103,6 @@ public class HoodieSparkCopyOnWriteTable<T extends HoodieRecordPayload>
super(config, context, metaClient);
}
- @Override
- public boolean isTableServiceAction(String actionType) {
- return !actionType.equals(HoodieTimeline.COMMIT_ACTION);
- }
-
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> upsert(HoodieEngineContext context, String instantTime, HoodieData<HoodieRecord<T>> records) {
return new SparkUpsertCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, records).execute();
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java
index efc667af297..c9d74246319 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java
@@ -79,11 +79,6 @@ public class HoodieSparkMergeOnReadTable<T extends HoodieRecordPayload> extends
super(config, context, metaClient);
}
- @Override
- public boolean isTableServiceAction(String actionType) {
- return !actionType.equals(HoodieTimeline.DELTA_COMMIT_ACTION);
- }
-
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> upsert(HoodieEngineContext context, String instantTime, HoodieData<HoodieRecord<T>> records) {
return new SparkUpsertDeltaCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, records).execute();
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
index 76b51e6970b..72aba1f1638 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
@@ -281,6 +281,29 @@ public class TestHoodieTimelineArchiver extends HoodieClientTestHarness {
}
}
+ @Test
+ public void testArchiveTableWithReplaceCommits() throws Exception {
+ HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(true, 2, 4, 2);
+ for (int i = 1; i < 7; i++) {
+ if (i < 3) {
+ testTable.doWriteOperation("0000000" + i, WriteOperationType.UPSERT, i == 1 ? Arrays.asList("p1", "p2") : Collections.emptyList(),
+ Arrays.asList("p1", "p2"), 2);
+ } else {
+ testTable.doWriteOperation("0000000" + i, WriteOperationType.INSERT_OVERWRITE, Collections.emptyList(), Arrays.asList("p1", "p2"), 2);
+ }
+ // trigger archival
+ Pair<List<HoodieInstant>, List<HoodieInstant>> commitsList = archiveAndGetCommitsList(writeConfig);
+ List<HoodieInstant> originalCommits = commitsList.getKey();
+ List<HoodieInstant> commitsAfterArchival = commitsList.getValue();
+
+ if (i == 6) {
+ // after all rounds, only 3 should be left in active timeline. 4,5,6
+ assertEquals(originalCommits, commitsAfterArchival);
+ assertEquals(3, originalCommits.size());
+ }
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testSavepointWithArchival(boolean archiveBeyondSavepoint) throws Exception {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java
index 8b1cb875c09..bd29e2d6a2f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java
@@ -68,7 +68,7 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> {
// Committed instant
COMPLETED,
// Invalid instant
- INVALID
+ NIL
}
private State state = State.COMPLETED;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
index e40341a2033..b418c89f949 100755
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
@@ -663,7 +663,7 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
List<HoodieInstant> allInstants = new ArrayList<>();
long instantTime = 1;
for (State state : State.values()) {
- if (state == State.INVALID) {
+ if (state == State.NIL) {
continue;
}
for (String action : HoodieTimeline.VALID_ACTIONS_IN_TIMELINE) {