You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "nsivabalan (via GitHub)" <gi...@apache.org> on 2023/08/05 21:22:07 UTC
[GitHub] [hudi] nsivabalan commented on a diff in pull request #9371: [HUDI-6647] Expand Hudi Java Client Functionality
nsivabalan commented on code in PR #9371:
URL: https://github.com/apache/hudi/pull/9371#discussion_r1285121130
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -75,10 +83,13 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.GREATER_THAN;
import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.isIndexingCommit;
-public abstract class BaseHoodieTableServiceClient<O> extends BaseHoodieClient implements RunsTableService {
+public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieClient implements RunsTableService {
Review Comment:
can we add java docs for the class, also calling out what I, T, O generally refers to
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -285,7 +373,29 @@ public HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime) {
* Commit Log Compaction and track metrics.
*/
protected void completeLogCompaction(HoodieCommitMetadata metadata, HoodieTable table, String logCompactionCommitTime) {
- throw new UnsupportedOperationException("Log compaction is not supported yet.");
+ this.context.setJobStatus(this.getClass().getSimpleName(), "Collect log compaction write status and commit compaction");
+ List<HoodieWriteStat> writeStats = metadata.getWriteStats();
+ final HoodieInstant logCompactionInstant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.LOG_COMPACTION_ACTION, logCompactionCommitTime);
Review Comment:
handleWriteErrors(writeStats, TableServiceType.LOG_COMPACT); ?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -146,7 +157,10 @@ protected void setPendingInflightAndRequestedInstants(Set<String> pendingInfligh
* @param metadata commit metadata for which pre commit is being invoked.
*/
protected void preCommit(HoodieCommitMetadata metadata) {
- // To be overridden by specific engines to perform conflict resolution if any.
+ // Create a Hoodie table after startTxn which encapsulated the commits and files visible.
Review Comment:
do we even need to have this protected now that no engine is overriding it
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -244,12 +305,39 @@ public Option<String> scheduleCompaction(Option<Map<String, String>> extraMetada
* @param metadata All the metadata that gets stored along with a commit
* @param extraMetadata Extra Metadata to be stored
*/
- public abstract void commitCompaction(String compactionInstantTime, HoodieCommitMetadata metadata, Option<Map<String, String>> extraMetadata);
+ public void commitCompaction(String compactionInstantTime, HoodieCommitMetadata metadata, Option<Map<String, String>> extraMetadata) {
+ extraMetadata.ifPresent(m -> m.forEach(metadata::addMetadata));
+ completeCompaction(metadata, createTable(config, context.getHadoopConf().get()), compactionInstantTime);
+ }
/**
* Commit Compaction and track metrics.
*/
- protected abstract void completeCompaction(HoodieCommitMetadata metadata, HoodieTable table, String compactionCommitTime);
+ protected void completeCompaction(HoodieCommitMetadata metadata, HoodieTable table, String compactionCommitTime) {
+ this.context.setJobStatus(this.getClass().getSimpleName(), "Collect compaction write status and commit compaction: " + config.getTableName());
+ List<HoodieWriteStat> writeStats = metadata.getWriteStats();
+ handleWriteErrors(writeStats, TableServiceType.COMPACT);
+ final HoodieInstant compactionInstant = HoodieTimeline.getCompactionInflightInstant(compactionCommitTime);
+ try {
+ this.txnManager.beginTransaction(Option.of(compactionInstant), Option.empty());
+ finalizeWrite(table, compactionCommitTime, writeStats);
+ // commit to data table after committing to metadata table.
+ updateTableMetadata(table, metadata, compactionInstant, context.emptyHoodieData());
Review Comment:
again, in flink this was named as writeTableMetadata, while in spark its updateTableMetadata and may be there are reasons for it. can we ensure we don't change anything for flink.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -974,4 +1159,20 @@ public void close() {
// Stop timeline-server if running
super.close();
}
+
+ protected void updateTableMetadata(HoodieTable table, HoodieCommitMetadata commitMetadata,
+ HoodieInstant hoodieInstant,
+ HoodieData<WriteStatus> writeStatuses) {
+ // 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())
+ .ifPresent(writer -> ((HoodieTableMetadataWriter) writer).update(commitMetadata, writeStatuses, hoodieInstant.getTimestamp()));
+ }
+
+ protected void handleWriteErrors(List<HoodieWriteStat> writeStats, TableServiceType tableServiceType) {
Review Comment:
this was never invoked in FlinkTableServiceClient.
lets try to keep the functionality as is.
atleast lets override the method and make it no-op. lets not change any functionality un-intentionally
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java:
##########
@@ -243,7 +169,31 @@ public HoodieWriteMetadata<List<WriteStatus>> cluster(String clusteringInstant,
}
@Override
- protected HoodieTable<?, ?, ?, ?> createTable(HoodieWriteConfig config, Configuration hadoopConf) {
+ protected void validateClusteringCommit(HoodieWriteMetadata<List<WriteStatus>> clusteringMetadata, String clusteringCommitTime, HoodieTable table) {
Review Comment:
clustering is not supported in flink. we should just throw exception here
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -174,11 +188,43 @@ private void inlineCompaction(HoodieTable table, Option<Map<String, String>> ext
/**
* Ensures compaction instant is in expected state and performs Log Compaction for the workload stored in instant-time.s
*
- * @param compactionInstantTime Compaction Instant Time
+ * @param logCompactionInstantTime Compaction Instant Time
* @return Collection of Write Status
*/
- protected HoodieWriteMetadata<O> logCompact(String compactionInstantTime, boolean shouldComplete) {
- throw new UnsupportedOperationException("Log compaction is not supported yet.");
+ protected HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime, boolean shouldComplete) {
Review Comment:
can we revisit all access specifiers and remove protected if its not required. later if we want to override in any of the engines, we can make them protected.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -285,7 +373,29 @@ public HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime) {
* Commit Log Compaction and track metrics.
*/
protected void completeLogCompaction(HoodieCommitMetadata metadata, HoodieTable table, String logCompactionCommitTime) {
- throw new UnsupportedOperationException("Log compaction is not supported yet.");
+ this.context.setJobStatus(this.getClass().getSimpleName(), "Collect log compaction write status and commit compaction");
+ List<HoodieWriteStat> writeStats = metadata.getWriteStats();
+ final HoodieInstant logCompactionInstant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.LOG_COMPACTION_ACTION, logCompactionCommitTime);
+ try {
+ this.txnManager.beginTransaction(Option.of(logCompactionInstant), Option.empty());
+ preCommit(metadata);
+ finalizeWrite(table, logCompactionCommitTime, writeStats);
+ // commit to data table after committing to metadata table.
+ writeTableMetadata(table, logCompactionCommitTime, HoodieTimeline.LOG_COMPACTION_ACTION, metadata, context.emptyHoodieData());
Review Comment:
same here there are minor differences b/w spark and flink.
##########
hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java:
##########
@@ -42,27 +48,31 @@ protected HoodieJavaTableServiceClient(HoodieEngineContext context,
}
@Override
- protected HoodieWriteMetadata<List<WriteStatus>> compact(String compactionInstantTime, boolean shouldComplete) {
- throw new HoodieNotSupportedException("Compact is not supported in HoodieJavaTableServiceClient");
+ protected void validateClusteringCommit(HoodieWriteMetadata<List<WriteStatus>> clusteringMetadata, String clusteringCommitTime, HoodieTable table) {
Review Comment:
we can probably keep this in BaseHoodieTableServiceClient only and leave it as protected. so that if any engine implements cluster, can use this method to validate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org