You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by co...@apache.org on 2022/12/13 11:37:27 UTC
[hudi] branch release-0.12.2 updated (737811f69c0 -> b9109cd9fad)
This is an automated email from the ASF dual-hosted git repository.
codope pushed a change to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
from 737811f69c0 [HUDI-5304] Disabling spark-sql core flow tests to unblock CI (#7346)
new d228a10770a [HUDI-5306] Unify RecordIterator and HoodieParquetReader with ClosableIterator (#7340)
new 12ef3127309 Revert "[MINOR] Bumping Azure Ubuntu image to 22.04, as 18.04 will be deprecated soon (#7347)" (#7350)
new 09aa8bc7dbd [HUDI-5302] Fix: compute hash key from recordKey failed when recordKeyValue contains ',' (#7342)
new 93d766df0c4 [MINOR] Disable the `SparkSqlCoreFlow` tests (#7368)
new 1e8315a4ce4 [HUDI-5179] Updated Hudi Release guide (#7212)
new 2489409a233 [HUDI-5331] Add schema settings with stream api (#7384)
new 421f1cd6b75 [HUDI-5294] Support type change for schema on read + reconcile schema (#7326)
new 49bc2eb6c63 [MINOR] Fix locale specific NumberFormatException in testutils HoodieTestDataGenerator (#7215)
new bc6d6aa35d3 [HUDI-5334] Fix checkpoint reading for structured streaming (#7389)
new bd60b7954fb [HUDI-3661] Flink async compaction is not thread safe when use watermark (#7399)
new 5c3aba37788 [HUDI-5290] Remove the lock in HoodieFlinkWriteClient#writeTableMetadata (#7320)
new 60168bff464 [HUDI-5314] add call help procedure (#7361)
new aef045b8ce7 [HUDI-5344] Fix CVE - upgrade protobuf-java to 3.18.2 (#6957)
new 8467e992ab0 [HUDI-4764] AWS GlueSync turn partition already exist error into warning (#6505)
new d3f420d9948 [HUDI-5163] Fix failure handling with spark datasource write (#7140)
new 77df347fc02 [HUDI-5344] Fix CVE - upgrade protobuf-java (#6960)
new b9109cd9fad [HUDI-5295] One meta sync failure should not prevent other meta sync from occurring (#7367)
The 17 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
azure-pipelines.yml | 2 +-
.../hudi/aws/sync/AWSGlueCatalogSyncClient.java | 6 +-
.../apache/hudi/client/BaseHoodieWriteClient.java | 2 +-
.../apache/hudi/index/bucket/BucketIdentifier.java | 9 +-
.../java/org/apache/hudi/keygen/KeyGenUtils.java | 39 +-
.../hudi/index/bucket/TestBucketIdentifier.java | 27 +-
.../org/apache/hudi/keygen/TestKeyGenUtils.java | 18 +
.../apache/hudi/client/HoodieFlinkWriteClient.java | 23 +-
.../org/apache/hudi/common/util/CommitUtils.java | 2 +-
.../hudi/common/util/InternalSchemaCache.java | 26 +-
.../schema/utils/AvroSchemaEvolutionUtils.java | 33 +-
.../common/testutils/HoodieTestDataGenerator.java | 3 +-
.../apache/hudi/configuration/OptionsResolver.java | 2 +-
.../hudi/sink/clustering/ClusteringOperator.java | 8 +-
.../{CompactFunction.java => CompactOperator.java} | 31 +-
.../hudi/sink/compact/CompactionCommitEvent.java | 2 +-
.../hudi/sink/compact/HoodieFlinkCompactor.java | 3 +-
.../java/org/apache/hudi/sink/utils/Pipelines.java | 5 +-
.../java/org/apache/hudi/table/format/CastMap.java | 19 +-
...Reader.java => ParquetSplitRecordIterator.java} | 24 +-
...odieParquetReader.java => RecordIterators.java} | 56 +-
...eader.java => SchemaEvolvedRecordIterator.java} | 25 +-
.../table/format/cow/CopyOnWriteInputFormat.java | 19 +-
.../table/format/mor/MergeOnReadInputFormat.java | 195 +++---
.../org/apache/hudi/util/FlinkWriteClients.java | 21 +-
.../java/org/apache/hudi/util/HoodiePipeline.java | 15 +
.../apache/hudi/sink/ITTestDataStreamWrite.java | 127 ++++
.../sink/TestStreamWriteOperatorCoordinator.java | 37 ++
.../sink/compact/ITTestHoodieFlinkCompactor.java | 5 +-
.../hudi/sink/utils/ClusteringFunctionWrapper.java | 2 +-
.../hudi/sink/utils/CompactFunctionWrapper.java | 53 +-
.../sink/utils/StreamWriteFunctionWrapper.java | 2 +-
.../main/scala/org/apache/hudi/DefaultSource.scala | 8 +-
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 46 +-
.../AlterHoodieTableDropPartitionCommand.scala | 6 +-
.../hudi/command/TruncateHoodieTableCommand.scala | 6 +-
hudi-spark-datasource/hudi-spark/pom.xml | 3 +
.../hudi/spark/sql/parser/HoodieSqlCommon.g4 | 6 +-
.../command/InsertIntoHoodieTableCommand.scala | 4 +
.../hudi/command/MergeIntoHoodieTableCommand.scala | 7 +-
.../hudi/command/procedures/HelpProcedure.scala | 125 ++++
.../hudi/command/procedures/HoodieProcedures.scala | 5 +
.../sql/parser/HoodieSqlCommonAstBuilder.scala | 21 +-
.../apache/hudi/functional/SparkSQLCoreFlow.java | 18 +-
.../hudi/functional/TestSparkSqlCoreFlow.scala | 3 +-
.../apache/spark/sql/hudi/TestInsertTable.scala | 52 +-
.../org/apache/spark/sql/hudi/TestSpark3DDL.scala | 106 +++-
.../sql/hudi/procedure/TestCommitsProcedure.scala | 2 +-
.../sql/hudi/procedure/TestHelpProcedure.scala | 84 +++
.../hudi/sync/common/util/SyncUtilHelpers.java | 14 +
.../hudi/utilities/deltastreamer/DeltaSync.java | 18 +-
.../deltastreamer/multisync/MockSyncTool1.java | 21 +-
.../deltastreamer/multisync/MockSyncTool2.java | 21 +-
.../multisync/MockSyncToolException1.java | 19 +-
.../multisync/MockSyncToolException2.java | 19 +-
.../multisync/TestMultipleMetaSync.java | 135 ++++
packaging/hudi-presto-bundle/pom.xml | 2 +-
packaging/hudi-trino-bundle/pom.xml | 2 +-
release/release_guide.md | 680 +++++++++++++++++++++
release/release_guide_overview.jpg | Bin 0 -> 363329 bytes
60 files changed, 1895 insertions(+), 379 deletions(-)
rename hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/{CompactFunction.java => CompactOperator.java} (82%)
rename hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/{HoodieParquetSplitReader.java => ParquetSplitRecordIterator.java} (63%)
rename hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/{HoodieParquetReader.java => RecordIterators.java} (67%)
rename hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/{HoodieParquetEvolvedSplitReader.java => SchemaEvolvedRecordIterator.java} (63%)
create mode 100644 hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HelpProcedure.scala
copy hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/providers/HoodieWriteClientProvider.java => hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/SparkSQLCoreFlow.java (68%)
create mode 100644 hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHelpProcedure.scala
copy hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/RandomFileIdPrefixProvider.java => hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool1.java (65%)
copy hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/RandomFileIdPrefixProvider.java => hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool2.java (65%)
copy hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/RandomFileIdPrefixProvider.java => hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException1.java (66%)
copy hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/RandomFileIdPrefixProvider.java => hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException2.java (66%)
create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/TestMultipleMetaSync.java
create mode 100644 release/release_guide.md
create mode 100644 release/release_guide_overview.jpg
[hudi] 02/17: Revert "[MINOR] Bumping Azure Ubuntu image to 22.04, as 18.04 will be deprecated soon (#7347)" (#7350)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 12ef312730935fc9823a781d5d3763029a2dd4c3
Author: Alexey Kudinkin <al...@gmail.com>
AuthorDate: Thu Dec 1 02:03:39 2022 -0800
Revert "[MINOR] Bumping Azure Ubuntu image to 22.04, as 18.04 will be deprecated soon (#7347)" (#7350)
This reverts commit eb6da963f9857329bf56b65c782357f032761cf2.
---
azure-pipelines.yml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index 17c2f6dd785..d450bfb9ab2 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -19,7 +19,7 @@ trigger:
- '*' # must quote since "*" is a YAML reserved character; we want a string
pool:
- vmImage: 'ubuntu-22.04'
+ vmImage: 'ubuntu-18.04'
parameters:
- name: job1Modules
[hudi] 17/17: [HUDI-5295] One meta sync failure should not prevent other meta sync from occurring (#7367)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit b9109cd9fad02910194fc8e10e78f5cd1fc4aeb7
Author: Jon Vexler <jb...@gmail.com>
AuthorDate: Wed Dec 7 22:23:34 2022 -0500
[HUDI-5295] One meta sync failure should not prevent other meta sync from occurring (#7367)
Co-authored-by: Jonathan Vexler <=>
---
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 14 ++-
.../hudi/sync/common/util/SyncUtilHelpers.java | 14 +++
.../hudi/utilities/deltastreamer/DeltaSync.java | 18 ++-
.../deltastreamer/multisync/MockSyncTool1.java | 39 ++++++
.../deltastreamer/multisync/MockSyncTool2.java | 39 ++++++
.../multisync/MockSyncToolException1.java | 37 ++++++
.../multisync/MockSyncToolException2.java | 37 ++++++
.../multisync/TestMultipleMetaSync.java | 135 +++++++++++++++++++++
8 files changed, 328 insertions(+), 5 deletions(-)
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 9442195c608..62c418e91f0 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
@@ -61,6 +61,7 @@ import org.apache.spark.{SPARK_VERSION, SparkContext}
import scala.collection.JavaConversions._
import scala.collection.mutable
+import scala.collection.mutable.ListBuffer
import scala.util.matching.Regex
object HoodieSparkSqlWriter {
@@ -711,9 +712,20 @@ object HoodieSparkSqlWriter {
properties.put(HoodieSyncConfig.META_SYNC_SPARK_VERSION.key, SPARK_VERSION)
properties.put(HoodieSyncConfig.META_SYNC_USE_FILE_LISTING_FROM_METADATA.key, hoodieConfig.getBoolean(HoodieMetadataConfig.ENABLE))
+ //Collect exceptions in list because we want all sync to run. Then we can throw
+ val metaSyncExceptions = new ListBuffer[HoodieException]()
syncClientToolClassSet.foreach(impl => {
- SyncUtilHelpers.runHoodieMetaSync(impl.trim, properties, fs.getConf, fs, basePath.toString, baseFileFormat)
+ try {
+ SyncUtilHelpers.runHoodieMetaSync(impl.trim, properties, fs.getConf, fs, basePath.toString, baseFileFormat)
+ } catch {
+ case e: HoodieException =>
+ log.info("SyncTool class " + impl.trim + " failed with exception", e)
+ metaSyncExceptions.add(e)
+ }
})
+ if (metaSyncExceptions.nonEmpty) {
+ throw SyncUtilHelpers.getExceptionFromList(metaSyncExceptions)
+ }
}
// Since Hive tables are now synced as Spark data source tables which are cached after Spark SQL queries
diff --git a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SyncUtilHelpers.java b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SyncUtilHelpers.java
index e7d61f96cff..26f90facdd3 100644
--- a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SyncUtilHelpers.java
+++ b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SyncUtilHelpers.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
+import java.util.Collection;
import java.util.Properties;
/**
@@ -104,4 +105,17 @@ public class SyncUtilHelpers {
+ ": no valid constructor found.");
}
}
+
+ public static HoodieException getExceptionFromList(Collection<HoodieException> exceptions) {
+ if (exceptions.size() == 1) {
+ return exceptions.stream().findFirst().get();
+ }
+ StringBuilder sb = new StringBuilder();
+ sb.append("Multiple exceptions during meta sync:\n");
+ exceptions.forEach(e -> {
+ sb.append(e.getMessage());
+ sb.append("\n");
+ });
+ return new HoodieException(sb.toString());
+ }
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
index 5b17a9f85be..9d8a5ef991b 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
@@ -743,11 +743,21 @@ public class DeltaSync implements Serializable {
props.getInteger(HoodieIndexConfig.BUCKET_INDEX_NUM_BUCKETS.key())));
}
+ //Collect exceptions in list because we want all sync to run. Then we can throw
+ List<HoodieException> metaSyncExceptions = new ArrayList<>();
for (String impl : syncClientToolClasses) {
- Timer.Context syncContext = metrics.getMetaSyncTimerContext();
- SyncUtilHelpers.runHoodieMetaSync(impl.trim(), metaProps, conf, fs, cfg.targetBasePath, cfg.baseFileFormat);
- long metaSyncTimeMs = syncContext != null ? syncContext.stop() : 0;
- metrics.updateDeltaStreamerMetaSyncMetrics(getSyncClassShortName(impl), metaSyncTimeMs);
+ try {
+ Timer.Context syncContext = metrics.getMetaSyncTimerContext();
+ SyncUtilHelpers.runHoodieMetaSync(impl.trim(), metaProps, conf, fs, cfg.targetBasePath, cfg.baseFileFormat);
+ long metaSyncTimeMs = syncContext != null ? syncContext.stop() : 0;
+ metrics.updateDeltaStreamerMetaSyncMetrics(getSyncClassShortName(impl), metaSyncTimeMs);
+ } catch (HoodieException e) {
+ LOG.info("SyncTool class " + impl.trim() + " failed with exception", e);
+ metaSyncExceptions.add(e);
+ }
+ }
+ if (!metaSyncExceptions.isEmpty()) {
+ throw SyncUtilHelpers.getExceptionFromList(metaSyncExceptions);
}
}
}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool1.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool1.java
new file mode 100644
index 00000000000..9ad2bd601ca
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool1.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.deltastreamer.multisync;
+
+import org.apache.hudi.sync.common.HoodieSyncTool;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.Properties;
+
+public class MockSyncTool1 extends HoodieSyncTool {
+
+ public static boolean syncSuccess;
+
+ public MockSyncTool1(Properties props, Configuration hadoopConf) {
+ super(props, hadoopConf);
+ }
+
+ @Override
+ public void syncHoodieTable() {
+ syncSuccess = true;
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool2.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool2.java
new file mode 100644
index 00000000000..68d85823c42
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncTool2.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.deltastreamer.multisync;
+
+import org.apache.hudi.sync.common.HoodieSyncTool;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.Properties;
+
+public class MockSyncTool2 extends HoodieSyncTool {
+
+ public static boolean syncSuccess;
+
+ public MockSyncTool2(Properties props, Configuration hadoopConf) {
+ super(props, hadoopConf);
+ }
+
+ @Override
+ public void syncHoodieTable() {
+ syncSuccess = true;
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException1.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException1.java
new file mode 100644
index 00000000000..1f8cd1a06fb
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException1.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.deltastreamer.multisync;
+
+import org.apache.hudi.sync.common.HoodieSyncTool;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.Properties;
+
+public class MockSyncToolException1 extends HoodieSyncTool {
+
+ public MockSyncToolException1(Properties props, Configuration hadoopConf) {
+ super(props, hadoopConf);
+ }
+
+ @Override
+ public void syncHoodieTable() {
+ throw new RuntimeException();
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException2.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException2.java
new file mode 100644
index 00000000000..6dcda819652
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/MockSyncToolException2.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.deltastreamer.multisync;
+
+import org.apache.hudi.sync.common.HoodieSyncTool;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.Properties;
+
+public class MockSyncToolException2 extends HoodieSyncTool {
+
+ public MockSyncToolException2(Properties props, Configuration hadoopConf) {
+ super(props, hadoopConf);
+ }
+
+ @Override
+ public void syncHoodieTable() {
+ throw new RuntimeException();
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/TestMultipleMetaSync.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/TestMultipleMetaSync.java
new file mode 100644
index 00000000000..3db11be49da
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/multisync/TestMultipleMetaSync.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.deltastreamer.multisync;
+
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer;
+import org.apache.hudi.utilities.functional.HoodieDeltaStreamerTestBase;
+import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
+import org.apache.hudi.utilities.sources.TestDataSource;
+import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
+import org.apache.hudi.utilities.transform.SqlQueryBasedTransformer;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestMultipleMetaSync extends HoodieDeltaStreamerTestBase {
+
+ @AfterAll
+ public static void cleanupClass() {
+ UtilitiesTestBase.cleanupClass();
+ if (testUtils != null) {
+ testUtils.teardown();
+ }
+ }
+
+ @Override
+ @BeforeEach
+ public void setup() throws Exception {
+ super.setup();
+ }
+
+ @Override
+ @AfterEach
+ public void teardown() throws Exception {
+ super.teardown();
+ }
+
+ @Test
+ void testMultipleMetaStore() throws Exception {
+ String tableBasePath = basePath + "/test_multiple_metastore";
+ MockSyncTool1.syncSuccess = false;
+ MockSyncTool2.syncSuccess = false;
+ // Initial bulk insert to ingest to first hudi table
+ HoodieDeltaStreamer.Config cfg = getConfig(tableBasePath, getSyncNames("MockSyncTool1", "MockSyncTool2"));
+ new HoodieDeltaStreamer(cfg, jsc, fs, hiveServer.getHiveConf()).sync();
+ assertTrue(MockSyncTool1.syncSuccess);
+ assertTrue(MockSyncTool2.syncSuccess);
+ }
+
+ @ParameterizedTest
+ @MethodSource("withOneException")
+ void testeWithException(String syncClassNames) {
+ String tableBasePath = basePath + "/test_multiple_metastore_exception";
+ MockSyncTool1.syncSuccess = false;
+ MockSyncTool2.syncSuccess = false;
+ HoodieDeltaStreamer.Config cfg = getConfig(tableBasePath, syncClassNames);
+ Exception e = assertThrows(HoodieException.class, () -> new HoodieDeltaStreamer(cfg, jsc, fs, hiveServer.getHiveConf()).sync());
+ assertTrue(e.getMessage().contains(MockSyncToolException1.class.getName()));
+ assertTrue(MockSyncTool1.syncSuccess);
+ assertTrue(MockSyncTool2.syncSuccess);
+ }
+
+ @Test
+ void testMultipleExceptions() {
+ String tableBasePath = basePath + "/test_multiple_metastore_multiple_exception";
+ MockSyncTool1.syncSuccess = false;
+ MockSyncTool2.syncSuccess = false;
+ HoodieDeltaStreamer.Config cfg = getConfig(tableBasePath, getSyncNames("MockSyncTool1", "MockSyncTool2", "MockSyncToolException1", "MockSyncToolException2"));
+ Exception e = assertThrows(HoodieException.class, () -> new HoodieDeltaStreamer(cfg, jsc, fs, hiveServer.getHiveConf()).sync());
+ assertTrue(e.getMessage().contains(MockSyncToolException1.class.getName()));
+ assertTrue(e.getMessage().contains(MockSyncToolException2.class.getName()));
+ assertTrue(MockSyncTool1.syncSuccess);
+ assertTrue(MockSyncTool2.syncSuccess);
+ }
+
+ HoodieDeltaStreamer.Config getConfig(String basePath, String syncClassNames) {
+ HoodieDeltaStreamer.Config cfg = new HoodieDeltaStreamer.Config();
+ cfg.targetBasePath = basePath;
+ cfg.targetTableName = "hoodie_trips";
+ cfg.tableType = "COPY_ON_WRITE";
+ cfg.sourceClassName = TestDataSource.class.getName();
+ cfg.transformerClassNames = Collections.singletonList(SqlQueryBasedTransformer.class.getName());
+ cfg.schemaProviderClassName = FilebasedSchemaProvider.class.getName();
+ cfg.syncClientToolClassNames = syncClassNames;
+ cfg.operation = WriteOperationType.BULK_INSERT;
+ cfg.enableHiveSync = true;
+ cfg.sourceOrderingField = "timestamp";
+ cfg.propsFilePath = UtilitiesTestBase.basePath + "/test-source.properties";
+ cfg.configs.add("hoodie.datasource.hive_sync.partition_fields=year,month,day");
+ cfg.sourceLimit = 1000;
+ return cfg;
+ }
+
+ private static String getSyncNames(String... syncs) {
+ return Arrays.stream(syncs).map(s -> "org.apache.hudi.utilities.deltastreamer.multisync." + s).collect(Collectors.joining(","));
+ }
+
+ private static Stream<Arguments> withOneException() {
+ return Stream.of(
+ Arguments.of(getSyncNames("MockSyncTool1", "MockSyncTool2", "MockSyncToolException1")),
+ Arguments.of(getSyncNames("MockSyncTool1", "MockSyncToolException1", "MockSyncTool2")),
+ Arguments.of(getSyncNames("MockSyncToolException1", "MockSyncTool1", "MockSyncTool2"))
+ );
+ }
+}
[hudi] 06/17: [HUDI-5331] Add schema settings with stream api (#7384)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 2489409a2337ca2909075330953e61e591473054
Author: superche <73...@users.noreply.github.com>
AuthorDate: Tue Dec 6 18:20:47 2022 +0800
[HUDI-5331] Add schema settings with stream api (#7384)
* Add schema set with stream api.
Co-authored-by: superche <su...@tencent.com>
---
.../java/org/apache/hudi/util/HoodiePipeline.java | 15 +++
.../apache/hudi/sink/ITTestDataStreamWrite.java | 127 +++++++++++++++++++++
2 files changed, 142 insertions(+)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/HoodiePipeline.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/HoodiePipeline.java
index f95367c8361..61ec2fccaa0 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/HoodiePipeline.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/HoodiePipeline.java
@@ -18,6 +18,8 @@
package org.apache.hudi.util;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.utils.EncodingUtils;
import org.apache.hudi.adapter.Utils;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.HoodieTableFactory;
@@ -125,6 +127,19 @@ public class HoodiePipeline {
return this;
}
+ /**
+ * Add table schema.
+ */
+ public Builder schema(Schema schema) {
+ for (Schema.UnresolvedColumn column : schema.getColumns()) {
+ column(column.toString());
+ }
+ if (schema.getPrimaryKey().isPresent()) {
+ pk(schema.getPrimaryKey().get().getColumnNames().stream().map(EncodingUtils::escapeIdentifier).collect(Collectors.joining(", ")));
+ }
+ return this;
+ }
+
/**
* Add a config option.
*/
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java
index 6ab4b1b6e0d..ad5642924da 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java
@@ -22,10 +22,13 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.configuration.OptionsInference;
import org.apache.hudi.sink.transform.ChainedTransformer;
import org.apache.hudi.sink.transform.Transformer;
import org.apache.hudi.sink.utils.Pipelines;
+import org.apache.hudi.table.catalog.HoodieCatalog;
+import org.apache.hudi.table.catalog.TableOptionProperties;
import org.apache.hudi.util.AvroSchemaConverter;
import org.apache.hudi.util.HoodiePipeline;
import org.apache.hudi.util.StreamerUtil;
@@ -49,6 +52,10 @@ import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
@@ -70,6 +77,9 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
+import static org.apache.hudi.table.catalog.CatalogOptions.CATALOG_PATH;
+import static org.apache.hudi.table.catalog.CatalogOptions.DEFAULT_DATABASE;
+
/**
* Integration test for Flink Hoodie stream sink.
*/
@@ -427,4 +437,121 @@ public class ITTestDataStreamWrite extends TestLogger {
execute(execEnv, false, "Api_Sink_Test");
TestData.checkWrittenDataCOW(tempFile, EXPECTED);
}
+
+ @Test
+ public void testHoodiePipelineBuilderSourceWithSchemaSet() throws Exception {
+ //create a StreamExecutionEnvironment instance.
+ StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ execEnv.getConfig().disableObjectReuse();
+ execEnv.setParallelism(1);
+ // set up checkpoint interval
+ execEnv.enableCheckpointing(4000, CheckpointingMode.EXACTLY_ONCE);
+ execEnv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
+
+ // create table dir
+ final String dbName = DEFAULT_DATABASE.defaultValue();
+ final String tableName = "t1";
+ File testTable = new File(tempFile, dbName + Path.SEPARATOR + tableName);
+ testTable.mkdir();
+
+ Configuration conf = TestConfigurations.getDefaultConf(testTable.toURI().toString());
+ conf.setString(FlinkOptions.TABLE_NAME, tableName);
+ conf.setString(FlinkOptions.TABLE_TYPE, "MERGE_ON_READ");
+
+ // write 3 batches of data set
+ TestData.writeData(TestData.dataSetInsert(1, 2), conf);
+ TestData.writeData(TestData.dataSetInsert(3, 4), conf);
+ TestData.writeData(TestData.dataSetInsert(5, 6), conf);
+
+ String latestCommit = TestUtils.getLastCompleteInstant(testTable.toURI().toString());
+
+ Map<String, String> options = new HashMap<>();
+ options.put(FlinkOptions.PATH.key(), testTable.toURI().toString());
+ options.put(FlinkOptions.READ_START_COMMIT.key(), latestCommit);
+
+ // create hoodie catalog, in order to get the table schema
+ Configuration catalogConf = new Configuration();
+ catalogConf.setString(CATALOG_PATH.key(), tempFile.toURI().toString());
+ catalogConf.setString(DEFAULT_DATABASE.key(), DEFAULT_DATABASE.defaultValue());
+ HoodieCatalog catalog = new HoodieCatalog("hudi", catalogConf);
+ catalog.open();
+ // get hoodieTable
+ ObjectPath tablePath = new ObjectPath(dbName, tableName);
+ TableOptionProperties.createProperties(testTable.toURI().toString(), HadoopConfigurations.getHadoopConf(catalogConf), options);
+ CatalogBaseTable hoodieTable = catalog.getTable(tablePath);
+
+ //read a hoodie table use low-level source api.
+ HoodiePipeline.Builder builder = HoodiePipeline.builder("test_source")
+ .schema(hoodieTable.getUnresolvedSchema())
+ .pk("uuid")
+ .partition("partition")
+ .options(options);
+ DataStream<RowData> rowDataDataStream = builder.source(execEnv);
+ List<RowData> result = new ArrayList<>();
+ rowDataDataStream.executeAndCollect().forEachRemaining(result::add);
+ TimeUnit.SECONDS.sleep(2);//sleep 2 second for collect data
+ TestData.assertRowDataEquals(result, TestData.dataSetInsert(5, 6));
+ }
+
+ @Test
+ public void testHoodiePipelineBuilderSinkWithSchemaSet() throws Exception {
+ StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ Map<String, String> options = new HashMap<>();
+ execEnv.getConfig().disableObjectReuse();
+ execEnv.setParallelism(4);
+ // set up checkpoint interval
+ execEnv.enableCheckpointing(4000, CheckpointingMode.EXACTLY_ONCE);
+ execEnv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
+
+ options.put(FlinkOptions.PATH.key(), tempFile.toURI().toString());
+ options.put(FlinkOptions.SOURCE_AVRO_SCHEMA_PATH.key(), Objects.requireNonNull(Thread.currentThread().getContextClassLoader().getResource("test_read_schema.avsc")).toString());
+ Configuration conf = Configuration.fromMap(options);
+ // Read from file source
+ RowType rowType =
+ (RowType) AvroSchemaConverter.convertToDataType(StreamerUtil.getSourceSchema(conf))
+ .getLogicalType();
+
+ JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema(
+ rowType,
+ InternalTypeInfo.of(rowType),
+ false,
+ true,
+ TimestampFormat.ISO_8601
+ );
+ String sourcePath = Objects.requireNonNull(Thread.currentThread()
+ .getContextClassLoader().getResource("test_source.data")).toString();
+
+ TextInputFormat format = new TextInputFormat(new Path(sourcePath));
+ format.setFilesFilter(FilePathFilter.createDefaultFilter());
+ format.setCharsetName("UTF-8");
+
+ DataStream dataStream = execEnv
+ // use continuous file source to trigger checkpoint
+ .addSource(new ContinuousFileSource.BoundedSourceFunction(new Path(sourcePath), 2))
+ .name("continuous_file_source")
+ .setParallelism(1)
+ .map(record -> deserializationSchema.deserialize(record.getBytes(StandardCharsets.UTF_8)))
+ .setParallelism(4);
+
+ Schema schema =
+ Schema.newBuilder()
+ .column("uuid", DataTypes.STRING().notNull())
+ .column("name", DataTypes.STRING())
+ .column("age", DataTypes.INT())
+ .column("ts", DataTypes.TIMESTAMP(3))
+ .column("partition", DataTypes.STRING())
+ .primaryKey("uuid")
+ .build();
+
+ //sink to hoodie table use low-level sink api.
+ HoodiePipeline.Builder builder = HoodiePipeline.builder("test_sink")
+ .schema(schema)
+ .partition("partition")
+ .options(options);
+
+ builder.sink(dataStream, false);
+
+ execute(execEnv, false, "Api_Sink_Test");
+ TestData.checkWrittenDataCOW(tempFile, EXPECTED);
+ }
}
[hudi] 12/17: [HUDI-5314] add call help procedure (#7361)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 60168bff464ebd1ad87dd745f8f52c5bb386fa8e
Author: 苏承祥 <sc...@aliyun.com>
AuthorDate: Wed Dec 7 20:03:30 2022 +0800
[HUDI-5314] add call help procedure (#7361)
* add call help procedure
Co-authored-by: 苏承祥 <su...@tuya.com>
---
.../hudi/spark/sql/parser/HoodieSqlCommon.g4 | 6 +-
.../hudi/command/procedures/HelpProcedure.scala | 125 +++++++++++++++++++++
.../hudi/command/procedures/HoodieProcedures.scala | 5 +
.../sql/parser/HoodieSqlCommonAstBuilder.scala | 21 ++--
.../sql/hudi/procedure/TestCommitsProcedure.scala | 2 +-
.../sql/hudi/procedure/TestHelpProcedure.scala | 84 ++++++++++++++
6 files changed, 231 insertions(+), 12 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/antlr4/org/apache/hudi/spark/sql/parser/HoodieSqlCommon.g4 b/hudi-spark-datasource/hudi-spark/src/main/antlr4/org/apache/hudi/spark/sql/parser/HoodieSqlCommon.g4
index 8643170f892..8a3106f7a56 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/antlr4/org/apache/hudi/spark/sql/parser/HoodieSqlCommon.g4
+++ b/hudi-spark-datasource/hudi-spark/src/main/antlr4/org/apache/hudi/spark/sql/parser/HoodieSqlCommon.g4
@@ -47,7 +47,7 @@
statement
: compactionStatement #compactionCommand
- | CALL multipartIdentifier '(' (callArgument (',' callArgument)*)? ')' #call
+ | CALL multipartIdentifier callArgumentList? #call
| CREATE INDEX (IF NOT EXISTS)? identifier ON TABLE?
tableIdentifier (USING indexType=identifier)?
LEFT_PAREN columns=multipartIdentifierPropertyList RIGHT_PAREN
@@ -69,6 +69,10 @@
: (db=IDENTIFIER '.')? table=IDENTIFIER
;
+ callArgumentList
+ : '(' (callArgument (',' callArgument)*)? ')'
+ ;
+
callArgument
: expression #positionalArgument
| identifier '=>' expression #namedArgument
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HelpProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HelpProcedure.scala
new file mode 100644
index 00000000000..b17d068e818
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HelpProcedure.scala
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.command.procedures
+
+import org.apache.hudi.exception.HoodieException
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+
+import java.util.function.Supplier
+
+class HelpProcedure extends BaseProcedure with ProcedureBuilder with Logging {
+
+ private val PARAMETERS = Array[ProcedureParameter](
+ ProcedureParameter.optional(0, "cmd", DataTypes.StringType, None)
+ )
+
+ private val OUTPUT_TYPE = new StructType(Array[StructField](
+ StructField("result", DataTypes.StringType, nullable = true, Metadata.empty)
+ ))
+
+
+ /**
+ * Returns the description of this procedure.
+ */
+ override def description: String = s"The procedure help command allows you to view all the commands currently provided, as well as their parameters and output fields."
+
+ def parameters: Array[ProcedureParameter] = PARAMETERS
+
+ def outputType: StructType = OUTPUT_TYPE
+
+ override def call(args: ProcedureArgs): Seq[Row] = {
+ super.checkArgs(PARAMETERS, args)
+ val line = "\n"
+ val tab = "\t"
+ if (args.map.isEmpty) {
+ val procedures: Map[String, Supplier[ProcedureBuilder]] = HoodieProcedures.procedures()
+ val result = new StringBuilder
+ result.append("synopsis").append(line)
+ .append(tab).append("call [command]([key1]=>[value1],[key2]=>[value2])").append(line)
+ result.append("commands and description").append(line)
+ procedures.keySet.foreach(name => {
+ val builderSupplier: Option[Supplier[ProcedureBuilder]] = procedures.get(name)
+ if (builderSupplier.isDefined) {
+ val procedure: Procedure = builderSupplier.get.get().build
+ result.append(tab)
+ .append(name).append(tab)
+ .append(procedure.description).append(line)
+ }
+ })
+ result.append("You can use 'call help(cmd=>[command])' to view the detailed parameters of the command").append(line)
+ Seq(Row(result.toString()))
+ } else {
+ val cmdOpt: Option[Any] = getArgValueOrDefault(args, PARAMETERS(0))
+ assert(cmdOpt.isDefined, "The cmd parameter is required")
+ val cmd: String = cmdOpt.get.asInstanceOf[String]
+ val procedures: Map[String, Supplier[ProcedureBuilder]] = HoodieProcedures.procedures()
+ val builderSupplier: Option[Supplier[ProcedureBuilder]] = procedures.get(cmd.trim)
+ if (builderSupplier.isEmpty) {
+ throw new HoodieException(s"can not find $cmd command in procedures.")
+ }
+ val procedure: Procedure = builderSupplier.get.get().build
+ val result = new StringBuilder
+
+ result.append("parameters:").append(line)
+ // set parameters header
+ result.append(tab)
+ .append(lengthFormat("param")).append(tab)
+ .append(lengthFormat("type_name")).append(tab)
+ .append(lengthFormat("default_value")).append(tab)
+ .append(lengthFormat("required")).append(line)
+ procedure.parameters.foreach(param => {
+ result.append(tab)
+ .append(lengthFormat(param.name)).append(tab)
+ .append(lengthFormat(param.dataType.typeName)).append(tab)
+ .append(lengthFormat(param.default.toString)).append(tab)
+ .append(lengthFormat(param.required.toString)).append(line)
+ })
+ result.append("outputType:").append(line)
+ // set outputType header
+ result.append(tab)
+ .append(lengthFormat("name")).append(tab)
+ .append(lengthFormat("type_name")).append(tab)
+ .append(lengthFormat("nullable")).append(tab)
+ .append(lengthFormat("metadata")).append(line)
+ procedure.outputType.map(field => {
+ result.append(tab)
+ .append(lengthFormat(field.name)).append(tab)
+ .append(lengthFormat(field.dataType.typeName)).append(tab)
+ .append(lengthFormat(field.nullable.toString)).append(tab)
+ .append(lengthFormat(field.metadata.toString())).append(line)
+ })
+ Seq(Row(result.toString()))
+ }
+ }
+
+ def lengthFormat(string: String): String = {
+ String.format("%-30s", string)
+ }
+
+ override def build = new HelpProcedure()
+}
+
+object HelpProcedure {
+ val NAME = "help"
+
+ def builder: Supplier[ProcedureBuilder] = new Supplier[ProcedureBuilder] {
+ override def get() = new HelpProcedure()
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
index 3b38925455f..a53fec33fe9 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
@@ -28,6 +28,10 @@ object HoodieProcedures {
if (builderSupplier.isDefined) builderSupplier.get.get() else null
}
+ def procedures(): Map[String, Supplier[ProcedureBuilder]] = {
+ BUILDERS
+ }
+
private def initProcedureBuilders: Map[String, Supplier[ProcedureBuilder]] = {
Map((RunCompactionProcedure.NAME, RunCompactionProcedure.builder)
,(ShowCompactionProcedure.NAME, ShowCompactionProcedure.builder)
@@ -83,6 +87,7 @@ object HoodieProcedures {
,(CopyToTempView.NAME, CopyToTempView.builder)
,(ShowCommitExtraMetadataProcedure.NAME, ShowCommitExtraMetadataProcedure.builder)
,(ShowTablePropertiesProcedure.NAME, ShowTablePropertiesProcedure.builder)
+ ,(HelpProcedure.NAME, HelpProcedure.builder)
)
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/parser/HoodieSqlCommonAstBuilder.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/parser/HoodieSqlCommonAstBuilder.scala
index d0e5ed61338..4005ef97e45 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/parser/HoodieSqlCommonAstBuilder.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/parser/HoodieSqlCommonAstBuilder.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedRelation}
import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
-import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface, ParserUtils}
+import org.apache.spark.sql.catalyst.parser.{ParserInterface, ParserUtils}
import org.apache.spark.sql.catalyst.plans.logical._
import java.util.Locale
@@ -92,13 +92,14 @@ class HoodieSqlCommonAstBuilder(session: SparkSession, delegate: ParserInterface
}
override def visitCall(ctx: CallContext): LogicalPlan = withOrigin(ctx) {
- if (ctx.callArgument().isEmpty) {
- throw new ParseException(s"Procedures arguments is empty", ctx)
+ if (ctx.callArgumentList() == null || ctx.callArgumentList().callArgument() == null || ctx.callArgumentList().callArgument().size() == 0) {
+ val name: Seq[String] = ctx.multipartIdentifier().parts.asScala.map(_.getText)
+ CallCommand(name, Seq())
+ } else {
+ val name: Seq[String] = ctx.multipartIdentifier().parts.asScala.map(_.getText)
+ val args: Seq[CallArgument] = ctx.callArgumentList().callArgument().asScala.map(typedVisit[CallArgument])
+ CallCommand(name, args)
}
-
- val name: Seq[String] = ctx.multipartIdentifier().parts.asScala.map(_.getText)
- val args: Seq[CallArgument] = ctx.callArgument().asScala.map(typedVisit[CallArgument])
- CallCommand(name, args)
}
/**
@@ -167,9 +168,9 @@ class HoodieSqlCommonAstBuilder(session: SparkSession, delegate: ParserInterface
}
val columns = ctx.columns.multipartIdentifierProperty.asScala
- .map(_.multipartIdentifier).map(typedVisit[Seq[String]]).toSeq
+ .map(_.multipartIdentifier).map(typedVisit[Seq[String]]).toSeq
val columnsProperties = ctx.columns.multipartIdentifierProperty.asScala
- .map(x => (Option(x.options).map(visitPropertyKeyValues).getOrElse(Map.empty))).toSeq
+ .map(x => (Option(x.options).map(visitPropertyKeyValues).getOrElse(Map.empty))).toSeq
val options = Option(ctx.indexOptions).map(visitPropertyKeyValues).getOrElse(Map.empty)
CreateIndex(
@@ -223,7 +224,7 @@ class HoodieSqlCommonAstBuilder(session: SparkSession, delegate: ParserInterface
* This should be called through [[visitPropertyKeyValues]] or [[visitPropertyKeys]].
*/
override def visitPropertyList(
- ctx: PropertyListContext): Map[String, String] = withOrigin(ctx) {
+ ctx: PropertyListContext): Map[String, String] = withOrigin(ctx) {
val properties = ctx.property.asScala.map { property =>
val key = visitPropertyKey(property.key)
val value = visitPropertyValue(property.value)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCommitsProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCommitsProcedure.scala
index 03cf26800df..febcd452799 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCommitsProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCommitsProcedure.scala
@@ -310,7 +310,7 @@ class TestCommitsProcedure extends HoodieSparkProcedureTestBase {
// Check required fields
checkExceptionContain(s"""call show_commit_extra_metadata()""")(
- s"arguments is empty")
+ s"Argument: table is required")
// collect commits for table
val commits = spark.sql(s"""call show_commits(table => '$tableName', limit => 10)""").collect()
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHelpProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHelpProcedure.scala
new file mode 100644
index 00000000000..2150682f893
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHelpProcedure.scala
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.procedure
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.hudi.command.procedures.{HoodieProcedures, Procedure, ProcedureBuilder}
+
+import java.util
+import java.util.function.Supplier
+
+class TestHelpProcedure extends HoodieSparkProcedureTestBase {
+
+ test("Test Call help Procedure with no params") {
+ val help: util.List[Row] = spark.sql("call help").collectAsList()
+ assert(help.size() == 1)
+
+ val help2: util.List[Row] = spark.sql("call help()").collectAsList()
+ assert(help2.size() == 1)
+
+ assert(help.get(0).toString().equals(help2.get(0).toString()))
+
+
+ val helpStr: String = help.get(0).toString()
+ val procedures: Map[String, Supplier[ProcedureBuilder]] = HoodieProcedures.procedures()
+
+ // check all procedures
+ procedures.keySet.foreach(name => {
+ // check cmd contains all procedure name
+ assert(helpStr.contains(name))
+ // check cmd contains all procedure description
+ val builderSupplier: Option[Supplier[ProcedureBuilder]] = procedures.get(name)
+ assert(builderSupplier.isDefined)
+ val procedure: Procedure = builderSupplier.get.get().build
+ assert(helpStr.contains(procedure.description))
+ })
+ }
+
+
+ test("Test Call help Procedure with params") {
+
+ // check not valid params
+ checkExceptionContain("call help(not_valid=>true)")("The cmd parameter is required")
+
+ val procedures: Map[String, Supplier[ProcedureBuilder]] = HoodieProcedures.procedures()
+
+ // check all procedures
+ procedures.keySet.foreach(name => {
+ val help: util.List[Row] = spark.sql(s"call help(cmd=>'$name')").collectAsList()
+ assert(help.size() == 1)
+
+ val helpStr: String = help.get(0).toString()
+ val builderSupplier: Option[Supplier[ProcedureBuilder]] = procedures.get(name)
+
+ assert(builderSupplier.isDefined)
+
+ // check result contains params
+ val procedure: Procedure = builderSupplier.get.get().build
+ procedure.parameters.foreach(params => {
+ assert(helpStr.contains(params.name))
+ })
+
+ // check result contains outputType
+ procedure.outputType.foreach(output => {
+ assert(helpStr.contains(output.name))
+ })
+ })
+ }
+
+}
[hudi] 14/17: [HUDI-4764] AWS GlueSync turn partition already exist error into warning (#6505)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 8467e992ab00ef44c09ba329f75132304e32b067
Author: Nicolas Paris <ni...@riseup.net>
AuthorDate: Wed Dec 7 14:00:56 2022 +0100
[HUDI-4764] AWS GlueSync turn partition already exist error into warning (#6505)
---
.../java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java b/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
index d9c4ecd96af..d0adbc9489a 100644
--- a/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
+++ b/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
@@ -138,8 +138,12 @@ public class AWSGlueCatalogSyncClient extends HoodieSyncClient {
BatchCreatePartitionResult result = awsGlue.batchCreatePartition(request);
if (CollectionUtils.nonEmpty(result.getErrors())) {
- throw new HoodieGlueSyncException("Fail to add partitions to " + tableId(databaseName, tableName)
+ if (result.getErrors().stream().allMatch((error) -> "AlreadyExistsException".equals(error.getErrorDetail().getErrorCode()))) {
+ LOG.warn("Partitions already exist in glue: " + result.getErrors());
+ } else {
+ throw new HoodieGlueSyncException("Fail to add partitions to " + tableId(databaseName, tableName)
+ " with error(s): " + result.getErrors());
+ }
}
Thread.sleep(BATCH_REQUEST_SLEEP_MILLIS);
}
[hudi] 11/17: [HUDI-5290] Remove the lock in HoodieFlinkWriteClient#writeTableMetadata (#7320)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 5c3aba37788e29a80f6b2e949318c77341e6a6e4
Author: just-JL <ji...@gmail.com>
AuthorDate: Wed Dec 7 19:45:49 2022 +0800
[HUDI-5290] Remove the lock in HoodieFlinkWriteClient#writeTableMetadata (#7320)
---
.../apache/hudi/client/HoodieFlinkWriteClient.java | 23 +++++---------
.../org/apache/hudi/util/FlinkWriteClients.java | 21 ++++++++----
.../sink/TestStreamWriteOperatorCoordinator.java | 37 ++++++++++++++++++++++
3 files changed, 59 insertions(+), 22 deletions(-)
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 12081e6bf1c..33148cddb11 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
@@ -282,21 +282,14 @@ public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
if (this.metadataWriter == null) {
initMetadataWriter();
}
- try {
- // guard the metadata writer with concurrent lock
- this.txnManager.getLockManager().lock();
-
- // refresh the timeline
-
- // Note: the data meta client is not refreshed currently, some code path
- // relies on the meta client for resolving the latest data schema,
- // 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));
- } finally {
- this.txnManager.getLockManager().unlock();
- }
+ // refresh the timeline
+
+ // Note: the data meta client is not refreshed currently, some code path
+ // relies on the meta client for resolving the latest data schema,
+ // 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));
}
/**
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java
index 41712e8fb98..ab409079c19 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java
@@ -26,6 +26,7 @@ import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.model.HoodieCleaningPolicy;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.config.HoodieArchivalConfig;
@@ -211,13 +212,6 @@ public class FlinkWriteClients {
.enable(conf.getBoolean(FlinkOptions.METADATA_ENABLED))
.withMaxNumDeltaCommitsBeforeCompaction(conf.getInteger(FlinkOptions.METADATA_COMPACTION_DELTA_COMMITS))
.build())
- .withLockConfig(HoodieLockConfig.newBuilder()
- .withLockProvider(FileSystemBasedLockProvider.class)
- .withLockWaitTimeInMillis(2000L) // 2s
- .withFileSystemLockExpire(1) // 1 minute
- .withClientNumRetries(30)
- .withFileSystemLockPath(StreamerUtil.getAuxiliaryPath(conf))
- .build())
.withPayloadConfig(getPayloadConfig(conf))
.withEmbeddedTimelineServerEnabled(enableEmbeddedTimelineService)
.withEmbeddedTimelineServerReuseEnabled(true) // make write client embedded timeline service singleton
@@ -226,6 +220,19 @@ public class FlinkWriteClients {
.withProps(flinkConf2TypedProperties(conf))
.withSchema(getSourceSchema(conf).toString());
+ if (conf.getBoolean(FlinkOptions.METADATA_ENABLED)) {
+ builder.withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL);
+ if (!conf.containsKey(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key())) {
+ builder.withLockConfig(HoodieLockConfig.newBuilder()
+ .withLockProvider(FileSystemBasedLockProvider.class)
+ .withLockWaitTimeInMillis(2000L) // 2s
+ .withFileSystemLockExpire(1) // 1 minute
+ .withClientNumRetries(30)
+ .withFileSystemLockPath(StreamerUtil.getAuxiliaryPath(conf))
+ .build());
+ }
+ }
+
// do not configure cleaning strategy as LAZY until multi-writers is supported.
HoodieWriteConfig writeConfig = builder.build();
if (loadFsViewStorageConfig && !conf.containsKey(FileSystemViewStorageConfig.REMOTE_HOST_NAME.key())) {
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
index 070937a1ab4..f2a8a49b5eb 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
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.config.HoodieWriteConfig;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.metadata.HoodieTableMetadata;
@@ -336,6 +337,42 @@ public class TestStreamWriteOperatorCoordinator {
}
}
+ @Test
+ void testLockForMetadataTable() throws Exception {
+ // reset
+ reset();
+ // override the default configuration
+ Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.setBoolean(FlinkOptions.METADATA_ENABLED, true);
+
+ conf.setString(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(), "optimistic_concurrency_control");
+ conf.setInteger("hoodie.write.lock.client.num_retries", 1);
+
+ OperatorCoordinator.Context context = new MockOperatorCoordinatorContext(new OperatorID(), 1);
+ coordinator = new StreamWriteOperatorCoordinator(conf, context);
+ coordinator.start();
+ coordinator.setExecutor(new MockCoordinatorExecutor(context));
+
+ final WriteMetadataEvent event0 = WriteMetadataEvent.emptyBootstrap(0);
+
+ coordinator.handleEventFromOperator(0, event0);
+
+ String instant = coordinator.getInstant();
+ assertNotEquals("", instant);
+
+ final String metadataTableBasePath = HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
+ HoodieTableMetaClient metadataTableMetaClient = StreamerUtil.createMetaClient(metadataTableBasePath, HadoopConfigurations.getHadoopConf(conf));
+ HoodieTimeline completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(1));
+ assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP));
+
+ instant = mockWriteWithMetadata();
+ metadataTableMetaClient.reloadActiveTimeline();
+ completedTimeline = metadataTableMetaClient.getActiveTimeline().filterCompletedInstants();
+ assertThat("One instant need to sync to metadata table", completedTimeline.countInstants(), is(2));
+ assertThat(completedTimeline.lastInstant().get().getTimestamp(), is(instant));
+ }
+
// -------------------------------------------------------------------------
// Utilities
// -------------------------------------------------------------------------
[hudi] 03/17: [HUDI-5302] Fix: compute hash key from recordKey failed when recordKeyValue contains ',' (#7342)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 09aa8bc7dbdd5390b3d1095d480ae822e658f6d5
Author: shaoxiong.zhan <31...@users.noreply.github.com>
AuthorDate: Sat Dec 3 10:47:22 2022 +0800
[HUDI-5302] Fix: compute hash key from recordKey failed when recordKeyValue contains ',' (#7342)
---
.../apache/hudi/index/bucket/BucketIdentifier.java | 9 +---
.../java/org/apache/hudi/keygen/KeyGenUtils.java | 39 +++++++++-------
.../hudi/index/bucket/TestBucketIdentifier.java | 27 ++++++-----
.../org/apache/hudi/keygen/TestKeyGenUtils.java | 18 ++++++++
.../apache/spark/sql/hudi/TestInsertTable.scala | 52 +++++++++++++++++++++-
5 files changed, 111 insertions(+), 34 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
index 48ccce1d174..35f9205a8e5 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
@@ -21,14 +21,13 @@ package org.apache.hudi.index.bucket;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.keygen.KeyGenUtils;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.regex.Pattern;
-import java.util.stream.Collectors;
public class BucketIdentifier implements Serializable {
// Compatible with the spark bucket name
@@ -69,11 +68,7 @@ public class BucketIdentifier implements Serializable {
}
private static List<String> getHashKeysUsingIndexFields(String recordKey, List<String> indexKeyFields) {
- Map<String, String> recordKeyPairs = Arrays.stream(recordKey.split(","))
- .map(p -> p.split(":"))
- .collect(Collectors.toMap(p -> p[0], p -> p[1]));
- return indexKeyFields.stream()
- .map(recordKeyPairs::get).collect(Collectors.toList());
+ return Arrays.asList(KeyGenUtils.extractRecordKeysByFields(recordKey, indexKeyFields));
}
public static String partitionBucketIdStr(String partition, int bucketId) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java
index d28263574b7..5a8b2b01c5f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java
@@ -33,6 +33,7 @@ import org.apache.hudi.keygen.parser.BaseHoodieDateTimeParser;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
public class KeyGenUtils {
@@ -43,6 +44,7 @@ public class KeyGenUtils {
protected static final String HUDI_DEFAULT_PARTITION_PATH = PartitionPathEncodeUtils.DEFAULT_PARTITION_PATH;
public static final String DEFAULT_PARTITION_PATH_SEPARATOR = "/";
public static final String DEFAULT_RECORD_KEY_PARTS_SEPARATOR = ",";
+ public static final String DEFAULT_COMPOSITE_KEY_FILED_VALUE = ":";
/**
* Fetches record key from the GenericRecord.
@@ -72,19 +74,24 @@ public class KeyGenUtils {
* @see org.apache.hudi.keygen.ComplexAvroKeyGenerator
*/
public static String[] extractRecordKeys(String recordKey) {
- String[] fieldKV = recordKey.split(",");
- return Arrays.stream(fieldKV).map(kv -> {
- final String[] kvArray = kv.split(":", 2);
- if (kvArray.length == 1) {
- return kvArray[0];
- } else if (kvArray[1].equals(NULL_RECORDKEY_PLACEHOLDER)) {
- return null;
- } else if (kvArray[1].equals(EMPTY_RECORDKEY_PLACEHOLDER)) {
- return "";
- } else {
- return kvArray[1];
- }
- }).toArray(String[]::new);
+ return extractRecordKeysByFields(recordKey, Collections.emptyList());
+ }
+
+ public static String[] extractRecordKeysByFields(String recordKey, List<String> fields) {
+ String[] fieldKV = recordKey.split(DEFAULT_RECORD_KEY_PARTS_SEPARATOR);
+ return Arrays.stream(fieldKV).map(kv -> kv.split(DEFAULT_COMPOSITE_KEY_FILED_VALUE, 2))
+ .filter(kvArray -> kvArray.length == 1 || fields.isEmpty() || (fields.contains(kvArray[0])))
+ .map(kvArray -> {
+ if (kvArray.length == 1) {
+ return kvArray[0];
+ } else if (kvArray[1].equals(NULL_RECORDKEY_PLACEHOLDER)) {
+ return null;
+ } else if (kvArray[1].equals(EMPTY_RECORDKEY_PLACEHOLDER)) {
+ return "";
+ } else {
+ return kvArray[1];
+ }
+ }).toArray(String[]::new);
}
public static String getRecordKey(GenericRecord record, List<String> recordKeyFields, boolean consistentLogicalTimestampEnabled) {
@@ -93,11 +100,11 @@ public class KeyGenUtils {
for (String recordKeyField : recordKeyFields) {
String recordKeyValue = HoodieAvroUtils.getNestedFieldValAsString(record, recordKeyField, true, consistentLogicalTimestampEnabled);
if (recordKeyValue == null) {
- recordKey.append(recordKeyField + ":" + NULL_RECORDKEY_PLACEHOLDER + ",");
+ recordKey.append(recordKeyField + DEFAULT_COMPOSITE_KEY_FILED_VALUE + NULL_RECORDKEY_PLACEHOLDER + DEFAULT_RECORD_KEY_PARTS_SEPARATOR);
} else if (recordKeyValue.isEmpty()) {
- recordKey.append(recordKeyField + ":" + EMPTY_RECORDKEY_PLACEHOLDER + ",");
+ recordKey.append(recordKeyField + DEFAULT_COMPOSITE_KEY_FILED_VALUE + EMPTY_RECORDKEY_PLACEHOLDER + DEFAULT_RECORD_KEY_PARTS_SEPARATOR);
} else {
- recordKey.append(recordKeyField + ":" + recordKeyValue + ",");
+ recordKey.append(recordKeyField + DEFAULT_COMPOSITE_KEY_FILED_VALUE + recordKeyValue + DEFAULT_RECORD_KEY_PARTS_SEPARATOR);
keyIsNullEmpty = false;
}
}
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestBucketIdentifier.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestBucketIdentifier.java
index 31f33890ad3..e6e6e4aabe7 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestBucketIdentifier.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestBucketIdentifier.java
@@ -26,12 +26,13 @@ import org.apache.hudi.keygen.KeyGenUtils;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
public class TestBucketIdentifier {
public static final String NESTED_COL_SCHEMA = "{\"type\":\"record\", \"name\":\"nested_col\",\"fields\": ["
@@ -103,20 +104,26 @@ public class TestBucketIdentifier {
public void testGetHashKeys() {
BucketIdentifier identifier = new BucketIdentifier();
List<String> keys = identifier.getHashKeys(new HoodieKey("abc", "partition"), "");
- Assertions.assertEquals(1, keys.size());
- Assertions.assertEquals("abc", keys.get(0));
+ assertEquals(1, keys.size());
+ assertEquals("abc", keys.get(0));
keys = identifier.getHashKeys(new HoodieKey("f1:abc", "partition"), "f1");
- Assertions.assertEquals(1, keys.size());
- Assertions.assertEquals("abc", keys.get(0));
+ assertEquals(1, keys.size());
+ assertEquals("abc", keys.get(0));
keys = identifier.getHashKeys(new HoodieKey("f1:abc,f2:bcd", "partition"), "f2");
- Assertions.assertEquals(1, keys.size());
- Assertions.assertEquals("bcd", keys.get(0));
+ assertEquals(1, keys.size());
+ assertEquals("bcd", keys.get(0));
keys = identifier.getHashKeys(new HoodieKey("f1:abc,f2:bcd", "partition"), "f1,f2");
- Assertions.assertEquals(2, keys.size());
- Assertions.assertEquals("abc", keys.get(0));
- Assertions.assertEquals("bcd", keys.get(1));
+ assertEquals(2, keys.size());
+ assertEquals("abc", keys.get(0));
+ assertEquals("bcd", keys.get(1));
+
+ keys = identifier.getHashKeys(new HoodieKey("f1:abc,f2:bcd,efg", "partition"), "f1,f2");
+ assertEquals(3, keys.size());
+ assertEquals("abc", keys.get(0));
+ assertEquals("bcd", keys.get(1));
+ assertEquals("efg", keys.get(2));
}
}
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java
index 43f5952e492..26e232ccc53 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java
@@ -21,6 +21,9 @@ package org.apache.hudi.keygen;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.List;
+
public class TestKeyGenUtils {
@Test
@@ -41,5 +44,20 @@ public class TestKeyGenUtils {
// test simple key form: val1
String[] s5 = KeyGenUtils.extractRecordKeys("1");
Assertions.assertArrayEquals(new String[] {"1"}, s5);
+
+ String[] s6 = KeyGenUtils.extractRecordKeys("id:1,id2:2,2");
+ Assertions.assertArrayEquals(new String[]{"1", "2", "2"}, s6);
+ }
+
+ @Test
+ public void testExtractRecordKeysWithFields() {
+ List<String> fields = new ArrayList<>(1);
+ fields.add("id2");
+
+ String[] s1 = KeyGenUtils.extractRecordKeysByFields("id1:1,id2:2,id3:3", fields);
+ Assertions.assertArrayEquals(new String[] {"2"}, s1);
+
+ String[] s2 = KeyGenUtils.extractRecordKeysByFields("id1:1,id2:2,2,id3:3", fields);
+ Assertions.assertArrayEquals(new String[] {"2", "2"}, s2);
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestInsertTable.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestInsertTable.scala
index c5e1abbf09a..2fa6b939acb 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestInsertTable.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestInsertTable.scala
@@ -24,7 +24,6 @@ import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.exception.HoodieDuplicateKeyException
import org.apache.hudi.keygen.ComplexKeyGenerator
import org.apache.spark.sql.SaveMode
-import org.apache.spark.sql.internal.SQLConf
import java.io.File
@@ -991,4 +990,55 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
spark.sql("set hoodie.merge.allow.duplicate.on.inserts = false")
spark.sql("set hoodie.datasource.write.operation = upsert")
}
+
+ test("Test Insert Into Bucket Index Table") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ // Create a partitioned table
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | dt string,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | tblproperties (
+ | primaryKey = 'id,name',
+ | preCombineField = 'ts',
+ | hoodie.index.type = 'BUCKET',
+ | hoodie.bucket.index.hash.field = 'id,name')
+ | partitioned by (dt)
+ | location '${tmp.getCanonicalPath}'
+ """.stripMargin)
+
+ // Note: Do not write the field alias, the partition field must be placed last.
+ spark.sql(
+ s"""
+ | insert into $tableName values
+ | (1, 'a1,1', 10, 1000, "2021-01-05"),
+ | (2, 'a2', 20, 2000, "2021-01-06"),
+ | (3, 'a3', 30, 3000, "2021-01-07")
+ """.stripMargin)
+
+ checkAnswer(s"select id, name, price, ts, dt from $tableName")(
+ Seq(1, "a1,1", 10.0, 1000, "2021-01-05"),
+ Seq(2, "a2", 20.0, 2000, "2021-01-06"),
+ Seq(3, "a3", 30.0, 3000, "2021-01-07")
+ )
+
+ spark.sql(
+ s"""
+ | insert into $tableName values
+ | (1, 'a1,1', 10, 1000, "2021-01-05")
+ """.stripMargin)
+
+ checkAnswer(s"select id, name, price, ts, dt from $tableName")(
+ Seq(1, "a1,1", 10.0, 1000, "2021-01-05"),
+ Seq(2, "a2", 20.0, 2000, "2021-01-06"),
+ Seq(3, "a3", 30.0, 3000, "2021-01-07")
+ )
+ }
+ }
}
[hudi] 07/17: [HUDI-5294] Support type change for schema on read + reconcile schema (#7326)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 421f1cd6b75b3a9599edd0953273f606ac112e3d
Author: xiarixiaoyao <me...@qq.com>
AuthorDate: Wed Dec 7 10:38:54 2022 +0800
[HUDI-5294] Support type change for schema on read + reconcile schema (#7326)
* [HUDI-5294] Support type change for schema on read + reconcile schema
---
.../apache/hudi/client/BaseHoodieWriteClient.java | 2 +-
.../hudi/common/util/InternalSchemaCache.java | 26 ++++-
.../schema/utils/AvroSchemaEvolutionUtils.java | 33 ++++---
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 32 +++++--
.../org/apache/spark/sql/hudi/TestSpark3DDL.scala | 106 ++++++++++++++++++++-
5 files changed, 172 insertions(+), 27 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..4cec5d1b0c3 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
@@ -293,7 +293,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
InternalSchema internalSchema;
Schema avroSchema = HoodieAvroUtils.createHoodieWriteSchema(config.getSchema(), config.allowOperationMetadataField());
if (historySchemaStr.isEmpty()) {
- internalSchema = AvroInternalSchemaConverter.convert(avroSchema);
+ internalSchema = SerDeHelper.fromJson(config.getInternalSchema()).orElse(AvroInternalSchemaConverter.convert(avroSchema));
internalSchema.setSchemaId(Long.parseLong(instantTime));
} else {
internalSchema = InternalSchemaUtils.searchSchema(Long.parseLong(instantTime),
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java b/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
index 846309b7b67..5f5a8763409 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
@@ -18,6 +18,8 @@
package org.apache.hudi.common.util;
+import org.apache.avro.Schema;
+import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -26,6 +28,7 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
import org.apache.hudi.internal.schema.io.FileBasedInternalSchemaStorageManager;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
@@ -166,9 +169,11 @@ public class InternalSchemaCache {
* step1：
* try to parser internalSchema from HoodieInstant directly
* step2：
- * if we cannot parser internalSchema in step1，
+ * if we cannot parser internalSchema in step1， (eg: current versionId HoodieInstant has been archived)
* try to find internalSchema in historySchema.
- *
+ * step3:
+ * if we cannot parser internalSchema in step2 (eg: schema evolution is not enabled when we create hoodie table, however after some inserts we enable schema evolution)
+ * try to convert table schema to internalSchema.
* @param versionId the internalSchema version to be search.
* @param tablePath table path
* @param hadoopConf conf
@@ -176,6 +181,7 @@ public class InternalSchemaCache {
* @return a internalSchema.
*/
public static InternalSchema getInternalSchemaByVersionId(long versionId, String tablePath, Configuration hadoopConf, String validCommits) {
+ String avroSchema = "";
Set<String> commitSet = Arrays.stream(validCommits.split(",")).collect(Collectors.toSet());
List<String> validateCommitList = commitSet.stream().map(fileName -> {
String fileExtension = HoodieInstant.getTimelineFileExtension(fileName);
@@ -199,6 +205,7 @@ public class InternalSchemaCache {
}
HoodieCommitMetadata metadata = HoodieCommitMetadata.fromBytes(data, HoodieCommitMetadata.class);
String latestInternalSchemaStr = metadata.getMetadata(SerDeHelper.LATEST_SCHEMA);
+ avroSchema = metadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY);
if (latestInternalSchemaStr != null) {
return SerDeHelper.fromJson(latestInternalSchemaStr).orElse(null);
}
@@ -209,8 +216,19 @@ public class InternalSchemaCache {
}
// step2:
FileBasedInternalSchemaStorageManager fileBasedInternalSchemaStorageManager = new FileBasedInternalSchemaStorageManager(hadoopConf, new Path(tablePath));
- String lastestHistorySchema = fileBasedInternalSchemaStorageManager.getHistorySchemaStrByGivenValidCommits(validateCommitList);
- return InternalSchemaUtils.searchSchema(versionId, SerDeHelper.parseSchemas(lastestHistorySchema));
+ String latestHistorySchema = fileBasedInternalSchemaStorageManager.getHistorySchemaStrByGivenValidCommits(validateCommitList);
+ if (latestHistorySchema.isEmpty()) {
+ return InternalSchema.getEmptyInternalSchema();
+ }
+ InternalSchema fileSchema = InternalSchemaUtils.searchSchema(versionId, SerDeHelper.parseSchemas(latestHistorySchema));
+ // step3:
+ return fileSchema.isEmptySchema() ? AvroInternalSchemaConverter.convert(HoodieAvroUtils.addMetadataFields(new Schema.Parser().parse(avroSchema))) : fileSchema;
+ }
+
+ public static InternalSchema getInternalSchemaByVersionId(long versionId, HoodieTableMetaClient metaClient) {
+ String validCommitLists = metaClient
+ .getCommitsAndCompactionTimeline().filterCompletedInstants().getInstantsAsStream().map(HoodieInstant::getFileName).collect(Collectors.joining(","));
+ return getInternalSchemaByVersionId(versionId, metaClient.getBasePathV2().toString(), metaClient.getHadoopConf(), validCommitLists);
}
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java b/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java
index e2b33915853..2561fc507e4 100644
--- a/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java
@@ -19,13 +19,11 @@
package org.apache.hudi.internal.schema.utils;
import org.apache.hudi.internal.schema.InternalSchema;
-import org.apache.hudi.internal.schema.Types;
import org.apache.hudi.internal.schema.action.TableChanges;
import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
import org.apache.avro.Schema;
-import java.util.ArrayList;
import java.util.List;
import java.util.TreeMap;
import java.util.stream.Collectors;
@@ -41,7 +39,8 @@ public class AvroSchemaEvolutionUtils {
* 2) incoming data contains new columns not defined yet in the table -> columns will be added to the table schema (incoming dataframe?)
* 3) incoming data has missing columns that are already defined in the table and new columns not yet defined in the table ->
* new columns will be added to the table schema, missing columns will be injected with null values
- * 4) support nested schema change.
+ * 4) support type change
+ * 5) support nested schema change.
* Notice:
* the incoming schema should not have delete/rename semantics.
* for example: incoming schema: int a, int b, int d; oldTableSchema int a, int b, int c, int d
@@ -52,25 +51,30 @@ public class AvroSchemaEvolutionUtils {
*/
public static InternalSchema reconcileSchema(Schema incomingSchema, InternalSchema oldTableSchema) {
InternalSchema inComingInternalSchema = AvroInternalSchemaConverter.convert(incomingSchema);
- // do check, only support add column evolution
+ // check column add/missing
List<String> colNamesFromIncoming = inComingInternalSchema.getAllColsFullName();
List<String> colNamesFromOldSchema = oldTableSchema.getAllColsFullName();
List<String> diffFromOldSchema = colNamesFromOldSchema.stream().filter(f -> !colNamesFromIncoming.contains(f)).collect(Collectors.toList());
- List<Types.Field> newFields = new ArrayList<>();
- if (colNamesFromIncoming.size() == colNamesFromOldSchema.size() && diffFromOldSchema.size() == 0) {
+ List<String> diffFromEvolutionColumns = colNamesFromIncoming.stream().filter(f -> !colNamesFromOldSchema.contains(f)).collect(Collectors.toList());
+ // check type change.
+ List<String> typeChangeColumns = colNamesFromIncoming
+ .stream()
+ .filter(f -> colNamesFromOldSchema.contains(f) && !inComingInternalSchema.findType(f).equals(oldTableSchema.findType(f)))
+ .collect(Collectors.toList());
+ if (colNamesFromIncoming.size() == colNamesFromOldSchema.size() && diffFromOldSchema.size() == 0 && typeChangeColumns.isEmpty()) {
return oldTableSchema;
}
- List<String> diffFromEvolutionSchema = colNamesFromIncoming.stream().filter(f -> !colNamesFromOldSchema.contains(f)).collect(Collectors.toList());
+
// Remove redundancy from diffFromEvolutionSchema.
// for example, now we add a struct col in evolvedSchema, the struct col is " user struct<name:string, age:int> "
// when we do diff operation: user, user.name, user.age will appeared in the resultSet which is redundancy, user.name and user.age should be excluded.
// deal with add operation
TreeMap<Integer, String> finalAddAction = new TreeMap<>();
- for (int i = 0; i < diffFromEvolutionSchema.size(); i++) {
- String name = diffFromEvolutionSchema.get(i);
+ for (int i = 0; i < diffFromEvolutionColumns.size(); i++) {
+ String name = diffFromEvolutionColumns.get(i);
int splitPoint = name.lastIndexOf(".");
String parentName = splitPoint > 0 ? name.substring(0, splitPoint) : "";
- if (!parentName.isEmpty() && diffFromEvolutionSchema.contains(parentName)) {
+ if (!parentName.isEmpty() && diffFromEvolutionColumns.contains(parentName)) {
// find redundancy, skip it
continue;
}
@@ -94,7 +98,14 @@ public class AvroSchemaEvolutionUtils {
inferPosition.map(i -> addChange.addPositionChange(name, i, "before"));
});
- return SchemaChangeUtils.applyTableChanges2Schema(oldTableSchema, addChange);
+ // do type evolution.
+ InternalSchema internalSchemaAfterAddColumns = SchemaChangeUtils.applyTableChanges2Schema(oldTableSchema, addChange);
+ TableChanges.ColumnUpdateChange typeChange = TableChanges.ColumnUpdateChange.get(internalSchemaAfterAddColumns);
+ typeChangeColumns.stream().filter(f -> !inComingInternalSchema.findType(f).isNestedType()).forEach(col -> {
+ typeChange.updateColumnType(col, inComingInternalSchema.findType(col));
+ });
+
+ return SchemaChangeUtils.applyTableChanges2Schema(internalSchemaAfterAddColumns, typeChange);
}
/**
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 0122576a679..9442195c608 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
@@ -258,8 +258,10 @@ object HoodieSparkSqlWriter {
if (reconcileSchema) {
// In case we need to reconcile the schema and schema evolution is enabled,
// we will force-apply schema evolution to the writer's schema
- if (schemaEvolutionEnabled && internalSchemaOpt.isEmpty) {
- internalSchemaOpt = Some(AvroInternalSchemaConverter.convert(sourceSchema))
+ if (schemaEvolutionEnabled) {
+ // in case sourceSchema contains HoodieRecord.HOODIE_META_COLUMNS
+ val allowOperationMetaDataField = parameters.getOrElse(HoodieWriteConfig.ALLOW_OPERATION_METADATA_FIELD.key(), "false").toBoolean
+ AvroInternalSchemaConverter.convert(HoodieAvroUtils.addMetadataFields(latestTableSchema, allowOperationMetaDataField))
}
if (internalSchemaOpt.isDefined) {
@@ -314,7 +316,7 @@ object HoodieSparkSqlWriter {
// Create a HoodieWriteClient & issue the write.
val client = hoodieWriteClient.getOrElse(DataSourceUtils.createHoodieClient(jsc, writerDataSchema.toString, path,
- tblName, mapAsJavaMap(addSchemaEvolutionParameters(parameters, internalSchemaOpt) - HoodieWriteConfig.AUTO_COMMIT_ENABLE.key)
+ tblName, mapAsJavaMap(addSchemaEvolutionParameters(parameters, internalSchemaOpt, Some(writerDataSchema)) - HoodieWriteConfig.AUTO_COMMIT_ENABLE.key)
)).asInstanceOf[SparkRDDWriteClient[HoodieRecordPayload[Nothing]]]
if (isAsyncCompactionEnabled(client, tableConfig, parameters, jsc.hadoopConfiguration())) {
@@ -399,10 +401,27 @@ object HoodieSparkSqlWriter {
processedRecord
}
- def addSchemaEvolutionParameters(parameters: Map[String, String], internalSchemaOpt: Option[InternalSchema]): Map[String, String] = {
+ def addSchemaEvolutionParameters(parameters: Map[String, String], internalSchemaOpt: Option[InternalSchema], writeSchemaOpt: Option[Schema] = None): Map[String, String] = {
val schemaEvolutionEnable = if (internalSchemaOpt.isDefined) "true" else "false"
- parameters ++ Map(HoodieWriteConfig.INTERNAL_SCHEMA_STRING.key() -> SerDeHelper.toJson(internalSchemaOpt.getOrElse(null)),
- HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key() -> schemaEvolutionEnable)
+
+ val schemaValidateEnable = if (schemaEvolutionEnable.toBoolean && parameters.getOrDefault(DataSourceWriteOptions.RECONCILE_SCHEMA.key(), "false").toBoolean) {
+ // force disable schema validate, now we support schema evolution, no need to do validate
+ "false"
+ } else {
+ parameters.getOrDefault(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key(), "true")
+ }
+ // correct internalSchema, internalSchema should contain hoodie metadata columns.
+ val correctInternalSchema = internalSchemaOpt.map { internalSchema =>
+ if (internalSchema.findField(HoodieRecord.RECORD_KEY_METADATA_FIELD) == null && writeSchemaOpt.isDefined) {
+ val allowOperationMetaDataField = parameters.getOrElse(HoodieWriteConfig.ALLOW_OPERATION_METADATA_FIELD.key(), "false").toBoolean
+ AvroInternalSchemaConverter.convert(HoodieAvroUtils.addMetadataFields(writeSchemaOpt.get, allowOperationMetaDataField))
+ } else {
+ internalSchema
+ }
+ }
+ parameters ++ Map(HoodieWriteConfig.INTERNAL_SCHEMA_STRING.key() -> SerDeHelper.toJson(correctInternalSchema.getOrElse(null)),
+ HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key() -> schemaEvolutionEnable,
+ HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key() -> schemaValidateEnable)
}
/**
@@ -411,7 +430,6 @@ object HoodieSparkSqlWriter {
* @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] = {
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..ed9db3a5aa4 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
@@ -18,13 +18,16 @@
package org.apache.spark.sql.hudi
import org.apache.hadoop.fs.Path
+import org.apache.hudi.DataSourceWriteOptions.{PARTITIONPATH_FIELD_OPT_KEY, PRECOMBINE_FIELD_OPT_KEY, RECORDKEY_FIELD_OPT_KEY, TABLE_NAME}
+import org.apache.hudi.QuickstartUtils.{DataGenerator, convertToStringList, getQuickstartWriteConfigs}
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.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.functions.{arrays_zip, col}
+import org.apache.spark.sql.functions.{arrays_zip, col, expr, lit}
+import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.{Row, SaveMode, SparkSession}
import scala.collection.JavaConversions._
@@ -511,7 +514,7 @@ class TestSpark3DDL extends HoodieSparkSqlTestBase {
}
}
- test("Test schema auto evolution") {
+ test("Test schema auto evolution complex") {
withTempDir { tmp =>
Seq("COPY_ON_WRITE", "MERGE_ON_READ").foreach { tableType =>
val tableName = generateTableName
@@ -534,7 +537,6 @@ class TestSpark3DDL extends HoodieSparkSqlTestBase {
DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY -> "partition",
DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY -> "timestamp",
"hoodie.schema.on.read.enable" -> "true",
- "hoodie.datasource.write.reconcile.schema" -> "true",
DataSourceWriteOptions.HIVE_STYLE_PARTITIONING_OPT_KEY -> "true"
)
@@ -546,7 +548,7 @@ class TestSpark3DDL extends HoodieSparkSqlTestBase {
.save(tablePath)
val oldView = spark.read.format("hudi").load(tablePath)
- oldView.show(false)
+ oldView.show(5, false)
val records2 = RawTripTestPayload.recordsToStrings(dataGen.generateUpdatesAsPerSchema("002", 100, schema)).toList
val inputD2 = spark.read.json(spark.sparkContext.parallelize(records2, 2))
@@ -571,4 +573,100 @@ class TestSpark3DDL extends HoodieSparkSqlTestBase {
}
}
}
+
+ test("Test schema auto evolution") {
+ withTempDir { tmp =>
+ Seq("COPY_ON_WRITE", "MERGE_ON_READ").foreach { tableType =>
+ // for complex schema.
+ val tableName = generateTableName
+ val tablePath = s"${new Path(tmp.getCanonicalPath, tableName).toUri.toString}"
+ if (HoodieSparkUtils.gteqSpark3_1) {
+ val dataGen = new DataGenerator
+ val inserts = convertToStringList(dataGen.generateInserts(10))
+ val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
+ df.write.format("hudi").
+ options(getQuickstartWriteConfigs).
+ option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, tableType).
+ option(PRECOMBINE_FIELD_OPT_KEY, "ts").
+ option(RECORDKEY_FIELD_OPT_KEY, "uuid").
+ option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
+ option("hoodie.schema.on.read.enable","true").
+ option(TABLE_NAME.key(), tableName).
+ option("hoodie.table.name", tableName).
+ mode("overwrite").
+ save(tablePath)
+
+ val updates = convertToStringList(dataGen.generateUpdates(10))
+ // type change: fare (double -> String)
+ // add new column and drop a column
+ val dfUpdate = spark.read.json(spark.sparkContext.parallelize(updates, 2))
+ .withColumn("fare", expr("cast(fare as string)"))
+ .withColumn("addColumn", lit("new"))
+ dfUpdate.drop("begin_lat").write.format("hudi").
+ options(getQuickstartWriteConfigs).
+ option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, tableType).
+ option(PRECOMBINE_FIELD_OPT_KEY, "ts").
+ option(RECORDKEY_FIELD_OPT_KEY, "uuid").
+ option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
+ option("hoodie.schema.on.read.enable","true").
+ option("hoodie.datasource.write.reconcile.schema","true").
+ option(TABLE_NAME.key(), tableName).
+ option("hoodie.table.name", tableName).
+ mode("append").
+ save(tablePath)
+ spark.sql("set hoodie.schema.on.read.enable=true")
+
+ val snapshotDF = spark.read.format("hudi").load(tablePath)
+
+ assertResult(StringType)(snapshotDF.schema.fields.filter(_.name == "fare").head.dataType)
+ assertResult("addColumn")(snapshotDF.schema.fields.last.name)
+ val checkRowKey = dfUpdate.select("fare").collectAsList().map(_.getString(0)).get(0)
+ snapshotDF.createOrReplaceTempView("hudi_trips_snapshot")
+ checkAnswer(spark.sql(s"select fare, addColumn from hudi_trips_snapshot where fare = ${checkRowKey}").collect())(
+ Seq(checkRowKey, "new")
+ )
+
+ spark.sql(s"select * from hudi_trips_snapshot").show(false)
+ // test insert_over_write + update again
+ val overwrite = convertToStringList(dataGen.generateInserts(10))
+ val dfOverWrite = spark.
+ read.json(spark.sparkContext.parallelize(overwrite, 2)).
+ filter("partitionpath = 'americas/united_states/san_francisco'")
+ .withColumn("fare", expr("cast(fare as string)")) // fare now in table is string type, we forbid convert string to double.
+ dfOverWrite.write.format("hudi").
+ options(getQuickstartWriteConfigs).
+ option("hoodie.datasource.write.operation","insert_overwrite").
+ option(PRECOMBINE_FIELD_OPT_KEY, "ts").
+ option(RECORDKEY_FIELD_OPT_KEY, "uuid").
+ option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
+ option("hoodie.schema.on.read.enable","true").
+ option("hoodie.datasource.write.reconcile.schema","true").
+ option(TABLE_NAME.key(), tableName).
+ option("hoodie.table.name", tableName).
+ mode("append").
+ save(tablePath)
+ spark.read.format("hudi").load(tablePath).show(false)
+
+ val updatesAgain = convertToStringList(dataGen.generateUpdates(10))
+ val dfAgain = spark.read.json(spark.sparkContext.parallelize(updatesAgain, 2)).withColumn("fare", expr("cast(fare as string)"))
+ dfAgain.write.format("hudi").
+ options(getQuickstartWriteConfigs).
+ option(PRECOMBINE_FIELD_OPT_KEY, "ts").
+ option(RECORDKEY_FIELD_OPT_KEY, "uuid").
+ option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
+ option("hoodie.schema.on.read.enable","true").
+ option("hoodie.datasource.write.reconcile.schema","true").
+ option(TABLE_NAME.key(), tableName).
+ option("hoodie.table.name", tableName).
+ mode("append").
+ save(tablePath)
+ spark.read.format("hudi").load(tablePath).createOrReplaceTempView("hudi_trips_snapshot1")
+ val checkKey = dfAgain.select("fare").collectAsList().map(_.getString(0)).get(0)
+ checkAnswer(spark.sql(s"select fare, addColumn from hudi_trips_snapshot1 where fare = ${checkKey}").collect())(
+ Seq(checkKey, null)
+ )
+ }
+ }
+ }
+ }
}
[hudi] 16/17: [HUDI-5344] Fix CVE - upgrade protobuf-java (#6960)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 77df347fc02dae21d0f6dfa3c3c516e236bd57a4
Author: lxxyyds <11...@users.noreply.github.com>
AuthorDate: Thu Dec 8 10:56:36 2022 +0800
[HUDI-5344] Fix CVE - upgrade protobuf-java (#6960)
* Fixes https://www.oscs1024.com/hd/CVE-2015-5237
Co-authored-by: Sagar Sumit <sa...@gmail.com>
---
packaging/hudi-presto-bundle/pom.xml | 2 +-
packaging/hudi-trino-bundle/pom.xml | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/packaging/hudi-presto-bundle/pom.xml b/packaging/hudi-presto-bundle/pom.xml
index 73d77ab6bb7..e7e59a1b85d 100644
--- a/packaging/hudi-presto-bundle/pom.xml
+++ b/packaging/hudi-presto-bundle/pom.xml
@@ -300,7 +300,7 @@
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
- <version>3.18.2</version>
+ <version>${proto.version}</version>
<scope>${presto.bundle.bootstrap.scope}</scope>
</dependency>
diff --git a/packaging/hudi-trino-bundle/pom.xml b/packaging/hudi-trino-bundle/pom.xml
index 1a8f58c7b36..4d273cb2087 100644
--- a/packaging/hudi-trino-bundle/pom.xml
+++ b/packaging/hudi-trino-bundle/pom.xml
@@ -288,7 +288,7 @@
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
- <version>2.5.0</version>
+ <version>${proto.version}</version>
<scope>${trino.bundle.bootstrap.scope}</scope>
</dependency>
[hudi] 10/17: [HUDI-3661] Flink async compaction is not thread safe when use watermark (#7399)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit bd60b7954fbeaf037241f96c01dcc42678180bb4
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Wed Dec 7 18:31:26 2022 +0800
[HUDI-3661] Flink async compaction is not thread safe when use watermark (#7399)
---
.../hudi/sink/clustering/ClusteringOperator.java | 8 +++-
.../{CompactFunction.java => CompactOperator.java} | 31 ++++++++++---
.../hudi/sink/compact/CompactionCommitEvent.java | 2 +-
.../hudi/sink/compact/HoodieFlinkCompactor.java | 3 +-
.../java/org/apache/hudi/sink/utils/Pipelines.java | 5 +-
.../sink/compact/ITTestHoodieFlinkCompactor.java | 5 +-
.../hudi/sink/utils/ClusteringFunctionWrapper.java | 2 +-
.../hudi/sink/utils/CompactFunctionWrapper.java | 53 +++++++++++-----------
.../sink/utils/StreamWriteFunctionWrapper.java | 2 +-
9 files changed, 66 insertions(+), 45 deletions(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
index ca1cd54c1fe..e7bde41ca8b 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
@@ -60,6 +60,7 @@ import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
@@ -159,7 +160,12 @@ public class ClusteringOperator extends TableStreamOperator<ClusteringCommitEven
this.executor = NonThrownExecutor.builder(LOG).build();
}
- collector = new StreamRecordCollector<>(output);
+ this.collector = new StreamRecordCollector<>(output);
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) {
+ // no need to propagate the watermark
}
@Override
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactOperator.java
similarity index 82%
rename from hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java
rename to hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactOperator.java
index 30c6d290b31..c5a59376b58 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactOperator.java
@@ -31,7 +31,11 @@ import org.apache.hudi.util.FlinkWriteClients;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.runtime.operators.TableStreamOperator;
+import org.apache.flink.table.runtime.util.StreamRecordCollector;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,11 +44,12 @@ import java.io.IOException;
import java.util.List;
/**
- * Function to execute the actual compaction task assigned by the compaction plan task.
+ * Operator to execute the actual compaction task assigned by the compaction plan task.
* In order to execute scalable, the input should shuffle by the compact event {@link CompactionPlanEvent}.
*/
-public class CompactFunction extends ProcessFunction<CompactionPlanEvent, CompactionCommitEvent> {
- private static final Logger LOG = LoggerFactory.getLogger(CompactFunction.class);
+public class CompactOperator extends TableStreamOperator<CompactionCommitEvent>
+ implements OneInputStreamOperator<CompactionPlanEvent, CompactionCommitEvent> {
+ private static final Logger LOG = LoggerFactory.getLogger(CompactOperator.class);
/**
* Config options.
@@ -71,22 +76,34 @@ public class CompactFunction extends ProcessFunction<CompactionPlanEvent, Compac
*/
private transient NonThrownExecutor executor;
- public CompactFunction(Configuration conf) {
+ /**
+ * Output records collector.
+ */
+ private transient StreamRecordCollector<CompactionCommitEvent> collector;
+
+ public CompactOperator(Configuration conf) {
this.conf = conf;
this.asyncCompaction = OptionsResolver.needsAsyncCompaction(conf);
}
@Override
- public void open(Configuration parameters) throws Exception {
+ public void open() throws Exception {
this.taskID = getRuntimeContext().getIndexOfThisSubtask();
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
if (this.asyncCompaction) {
this.executor = NonThrownExecutor.builder(LOG).build();
}
+ this.collector = new StreamRecordCollector<>(output);
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) {
+ // no need to propagate the watermark
}
@Override
- public void processElement(CompactionPlanEvent event, Context context, Collector<CompactionCommitEvent> collector) throws Exception {
+ public void processElement(StreamRecord<CompactionPlanEvent> record) throws Exception {
+ final CompactionPlanEvent event = record.getValue();
final String instantTime = event.getCompactionInstantTime();
final CompactionOperation compactionOperation = event.getOperation();
if (asyncCompaction) {
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java
index 398dfcf6195..faad4c2338d 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java
@@ -24,7 +24,7 @@ import java.io.Serializable;
import java.util.List;
/**
- * Represents a commit event from the compaction task {@link CompactFunction}.
+ * Represents a commit event from the compaction task {@link CompactOperator}.
*/
public class CompactionCommitEvent implements Serializable {
private static final long serialVersionUID = 1L;
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
index 0fb28eb86a8..b97306ff298 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
@@ -41,7 +41,6 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.client.deployment.application.ApplicationExecutionException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.operators.ProcessOperator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -312,7 +311,7 @@ public class HoodieFlinkCompactor {
.rebalance()
.transform("compact_task",
TypeInformation.of(CompactionCommitEvent.class),
- new ProcessOperator<>(new CompactFunction(conf)))
+ new CompactOperator(conf))
.setParallelism(compactionParallelism)
.addSink(new CompactionCommitSink(conf))
.name("compaction_commit")
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
index a045a9276c5..d17213dcc04 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
@@ -39,7 +39,7 @@ import org.apache.hudi.sink.clustering.ClusteringOperator;
import org.apache.hudi.sink.clustering.ClusteringPlanEvent;
import org.apache.hudi.sink.clustering.ClusteringPlanOperator;
import org.apache.hudi.sink.common.WriteOperatorFactory;
-import org.apache.hudi.sink.compact.CompactFunction;
+import org.apache.hudi.sink.compact.CompactOperator;
import org.apache.hudi.sink.compact.CompactionCommitEvent;
import org.apache.hudi.sink.compact.CompactionCommitSink;
import org.apache.hudi.sink.compact.CompactionPlanEvent;
@@ -57,7 +57,6 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
-import org.apache.flink.streaming.api.operators.ProcessOperator;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
@@ -372,7 +371,7 @@ public class Pipelines {
.keyBy(plan -> plan.getOperation().getFileGroupId().getFileId())
.transform("compact_task",
TypeInformation.of(CompactionCommitEvent.class),
- new ProcessOperator<>(new CompactFunction(conf)))
+ new CompactOperator(conf))
.setParallelism(conf.getInteger(FlinkOptions.COMPACTION_TASKS))
.addSink(new CompactionCommitSink(conf))
.name("compact_commit")
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
index 711b738288f..6157b5e9011 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
@@ -39,7 +39,6 @@ import org.apache.hudi.utils.TestSQL;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.operators.ProcessOperator;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
@@ -158,7 +157,7 @@ public class ITTestHoodieFlinkCompactor {
.rebalance()
.transform("compact_task",
TypeInformation.of(CompactionCommitEvent.class),
- new ProcessOperator<>(new CompactFunction(conf)))
+ new CompactOperator(conf))
.setParallelism(FlinkMiniCluster.DEFAULT_PARALLELISM)
.addSink(new CompactionCommitSink(conf))
.name("clean_commits")
@@ -282,7 +281,7 @@ public class ITTestHoodieFlinkCompactor {
.rebalance()
.transform("compact_task",
TypeInformation.of(CompactionCommitEvent.class),
- new ProcessOperator<>(new CompactFunction(conf)))
+ new CompactOperator(conf))
.setParallelism(1)
.addSink(new CompactionCommitSink(conf))
.name("compaction_commit")
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/ClusteringFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/ClusteringFunctionWrapper.java
index 55a79915d47..e3b75cbf637 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/ClusteringFunctionWrapper.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/ClusteringFunctionWrapper.java
@@ -69,6 +69,7 @@ public class ClusteringFunctionWrapper {
private ClusteringCommitSink commitSink;
public ClusteringFunctionWrapper(Configuration conf, StreamTask<?, ?> streamTask, StreamConfig streamConfig) {
+ this.conf = conf;
this.ioManager = new IOManagerAsync();
MockEnvironment environment = new MockEnvironmentBuilder()
.setTaskName("mockTask")
@@ -76,7 +77,6 @@ public class ClusteringFunctionWrapper {
.setIOManager(ioManager)
.build();
this.runtimeContext = new MockStreamingRuntimeContext(false, 1, 0, environment);
- this.conf = conf;
this.streamTask = streamTask;
this.streamConfig = streamConfig;
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/CompactFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/CompactFunctionWrapper.java
index 1dba81ce2b7..78a8305c9c5 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/CompactFunctionWrapper.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/CompactFunctionWrapper.java
@@ -19,7 +19,7 @@
package org.apache.hudi.sink.utils;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
-import org.apache.hudi.sink.compact.CompactFunction;
+import org.apache.hudi.sink.compact.CompactOperator;
import org.apache.hudi.sink.compact.CompactionCommitEvent;
import org.apache.hudi.sink.compact.CompactionCommitSink;
import org.apache.hudi.sink.compact.CompactionPlanEvent;
@@ -33,14 +33,14 @@ import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
+import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
-import org.apache.flink.util.Collector;
-
-import java.util.ArrayList;
-import java.util.List;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
/**
- * A wrapper class to manipulate the {@link org.apache.hudi.sink.compact.CompactFunction} instance for testing.
+ * A wrapper class to manipulate the {@link CompactOperator} instance for testing.
*/
public class CompactFunctionWrapper {
private final Configuration conf;
@@ -48,20 +48,28 @@ public class CompactFunctionWrapper {
private final IOManager ioManager;
private final StreamingRuntimeContext runtimeContext;
+ private final StreamTask<?, ?> streamTask;
+ private final StreamConfig streamConfig;
+
/**
* Function that generates the {@link HoodieCompactionPlan}.
*/
private CompactionPlanOperator compactionPlanOperator;
+ /**
+ * Output to collect the compaction commit events.
+ */
+ private CollectorOutput<CompactionCommitEvent> commitEventOutput;
/**
* Function that executes the compaction task.
*/
- private CompactFunction compactFunction;
+ private CompactOperator compactOperator;
/**
* Stream sink to handle compaction commits.
*/
private CompactionCommitSink commitSink;
- public CompactFunctionWrapper(Configuration conf) throws Exception {
+ public CompactFunctionWrapper(Configuration conf, StreamTask<?, ?> streamTask, StreamConfig streamConfig) {
+ this.conf = conf;
this.ioManager = new IOManagerAsync();
MockEnvironment environment = new MockEnvironmentBuilder()
.setTaskName("mockTask")
@@ -69,19 +77,23 @@ public class CompactFunctionWrapper {
.setIOManager(ioManager)
.build();
this.runtimeContext = new MockStreamingRuntimeContext(false, 1, 0, environment);
- this.conf = conf;
+ this.streamTask = streamTask;
+ this.streamConfig = streamConfig;
}
public void openFunction() throws Exception {
compactionPlanOperator = new CompactionPlanOperator(conf);
compactionPlanOperator.open();
- compactFunction = new CompactFunction(conf);
- compactFunction.setRuntimeContext(runtimeContext);
- compactFunction.open(conf);
+ compactOperator = new CompactOperator(conf);
+ // CAUTION: deprecated API used.
+ compactOperator.setProcessingTimeService(new TestProcessingTimeService());
+ commitEventOutput = new CollectorOutput<>();
+ compactOperator.setup(streamTask, streamConfig, commitEventOutput);
+ compactOperator.open();
final NonThrownExecutor syncExecutor = new MockCoordinatorExecutor(
new MockOperatorCoordinatorContext(new OperatorID(), 1));
- compactFunction.setExecutor(syncExecutor);
+ compactOperator.setExecutor(syncExecutor);
commitSink = new CompactionCommitSink(conf);
commitSink.setRuntimeContext(runtimeContext);
@@ -94,22 +106,11 @@ public class CompactFunctionWrapper {
compactionPlanOperator.setOutput(output);
compactionPlanOperator.notifyCheckpointComplete(checkpointID);
// collect the CompactCommitEvents
- List<CompactionCommitEvent> compactCommitEvents = new ArrayList<>();
for (CompactionPlanEvent event : output.getRecords()) {
- compactFunction.processElement(event, null, new Collector<CompactionCommitEvent>() {
- @Override
- public void collect(CompactionCommitEvent event) {
- compactCommitEvents.add(event);
- }
-
- @Override
- public void close() {
-
- }
- });
+ compactOperator.processElement(new StreamRecord<>(event));
}
// handle and commit the compaction
- for (CompactionCommitEvent event : compactCommitEvents) {
+ for (CompactionCommitEvent event : commitEventOutput.getRecords()) {
commitSink.invoke(event, null);
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java
index a1a14456e3c..db8ff36962b 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java
@@ -117,7 +117,6 @@ public class StreamWriteFunctionWrapper<I> implements TestFunctionWrapper<I> {
// one function
this.coordinatorContext = new MockOperatorCoordinatorContext(new OperatorID(), 1);
this.coordinator = new StreamWriteOperatorCoordinator(conf, this.coordinatorContext);
- this.compactFunctionWrapper = new CompactFunctionWrapper(this.conf);
this.bucketAssignFunctionContext = new MockBucketAssignFunctionContext();
this.stateInitializationContext = new MockStateInitializationContext();
this.asyncCompaction = OptionsResolver.needsAsyncCompaction(conf);
@@ -127,6 +126,7 @@ public class StreamWriteFunctionWrapper<I> implements TestFunctionWrapper<I> {
.setConfig(new StreamConfig(conf))
.setExecutionConfig(new ExecutionConfig().enableObjectReuse())
.build();
+ this.compactFunctionWrapper = new CompactFunctionWrapper(this.conf, this.streamTask, this.streamConfig);
}
public void openFunction() throws Exception {
[hudi] 15/17: [HUDI-5163] Fix failure handling with spark datasource write (#7140)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d3f420d99485ce4ce03b047ca074e086fb3875b0
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Wed Dec 7 09:17:27 2022 -0800
[HUDI-5163] Fix failure handling with spark datasource write (#7140)
---
.../src/main/scala/org/apache/hudi/DefaultSource.scala | 8 ++++++--
.../sql/hudi/command/AlterHoodieTableDropPartitionCommand.scala | 6 +++++-
.../spark/sql/hudi/command/TruncateHoodieTableCommand.scala | 6 +++++-
.../spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala | 4 ++++
.../spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala | 7 ++++++-
5 files changed, 26 insertions(+), 5 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala
index 5641a622cbc..160e71a6ec4 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala
@@ -140,11 +140,15 @@ class DefaultSource extends RelationProvider
if (optParams.get(OPERATION.key).contains(BOOTSTRAP_OPERATION_OPT_VAL)) {
HoodieSparkSqlWriter.bootstrap(sqlContext, mode, optParams, dfWithoutMetaCols)
+ HoodieSparkSqlWriter.cleanup()
} else {
- HoodieSparkSqlWriter.write(sqlContext, mode, optParams, dfWithoutMetaCols)
+ val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(sqlContext, mode, optParams, dfWithoutMetaCols)
+ HoodieSparkSqlWriter.cleanup()
+ if (!success) {
+ throw new HoodieException("Write to Hudi failed")
+ }
}
- HoodieSparkSqlWriter.cleanup()
new HoodieEmptyRelation(sqlContext, dfWithoutMetaCols.schema)
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterHoodieTableDropPartitionCommand.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterHoodieTableDropPartitionCommand.scala
index 628f383b690..c6aa2e7aeda 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterHoodieTableDropPartitionCommand.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterHoodieTableDropPartitionCommand.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.hudi.command
import org.apache.hudi.HoodieSparkSqlWriter
+import org.apache.hudi.exception.HoodieException
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable
@@ -59,11 +60,14 @@ case class AlterHoodieTableDropPartitionCommand(
// delete partition files by enabling cleaner and setting retention policies.
val partitionsToDrop = getPartitionPathToDrop(hoodieCatalogTable, normalizedSpecs)
val parameters = buildHoodieDropPartitionsConfig(sparkSession, hoodieCatalogTable, partitionsToDrop)
- HoodieSparkSqlWriter.write(
+ val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(
sparkSession.sqlContext,
SaveMode.Append,
parameters,
sparkSession.emptyDataFrame)
+ if (!success) {
+ throw new HoodieException("Alter table command failed")
+ }
sparkSession.catalog.refreshTable(tableIdentifier.unquotedString)
logInfo(s"Finish execute alter table drop partition command for $fullTableName")
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/TruncateHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/TruncateHoodieTableCommand.scala
index f5349ee5fee..05f96efdae5 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/TruncateHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/TruncateHoodieTableCommand.scala
@@ -22,6 +22,7 @@ import org.apache.hudi.HoodieSparkSqlWriter
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.exception.HoodieException
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.catalog.{CatalogTableType, HoodieCatalogTable}
@@ -85,11 +86,14 @@ case class TruncateHoodieTableCommand(
// drop partitions to lazy clean
val partitionsToDrop = getPartitionPathToDrop(hoodieCatalogTable, normalizedSpecs)
val parameters = buildHoodieDropPartitionsConfig(sparkSession, hoodieCatalogTable, partitionsToDrop)
- HoodieSparkSqlWriter.write(
+ val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(
sparkSession.sqlContext,
SaveMode.Append,
parameters,
sparkSession.emptyDataFrame)
+ if (!success) {
+ throw new HoodieException("Truncate Hoodie Table command failed")
+ }
}
// After deleting the data, refresh the table to make sure we don't keep around a stale
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
index 8bd81df3d27..125e8028020 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
@@ -100,6 +100,10 @@ object InsertIntoHoodieTableCommand extends Logging with ProvidesHoodieConfig wi
val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(sparkSession.sqlContext, mode, config, Dataset.ofRows(sparkSession, alignedQuery))
+ if (!success) {
+ throw new HoodieException("Insert Into to Hudi table failed")
+ }
+
if (success && refreshTable) {
sparkSession.catalog.refreshTable(table.identifier.unquotedString)
}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala
index 9919062cac7..b689d7b2b76 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala
@@ -22,6 +22,7 @@ import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.util.StringUtils
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.config.HoodieWriteConfig.TBL_NAME
+import org.apache.hudi.exception.HoodieException
import org.apache.hudi.hive.HiveSyncConfigHolder
import org.apache.hudi.sync.common.HoodieSyncConfig
import org.apache.hudi.{AvroConversionUtils, DataSourceWriteOptions, HoodieSparkSqlWriter, SparkAdapterSupport}
@@ -348,7 +349,11 @@ case class MergeIntoHoodieTableCommand(mergeInto: MergeIntoTable) extends Hoodie
// Remove the meta fields from the sourceDF as we do not need these when writing.
val sourceDFWithoutMetaFields = removeMetaFields(sourceDF)
- HoodieSparkSqlWriter.write(sparkSession.sqlContext, SaveMode.Append, writeParams, sourceDFWithoutMetaFields)
+
+ val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(sparkSession.sqlContext, SaveMode.Append, writeParams, sourceDFWithoutMetaFields)
+ if (!success) {
+ throw new HoodieException("Merge into Hoodie table command failed")
+ }
}
private def checkUpdateAssignments(updateActions: Seq[UpdateAction]): Unit = {
[hudi] 08/17: [MINOR] Fix locale specific NumberFormatException in testutils HoodieTestDataGenerator (#7215)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 49bc2eb6c63487a35bef9d3507bcef8ece2157c9
Author: Alexander Trushev <tr...@gmail.com>
AuthorDate: Wed Dec 7 13:42:14 2022 +0700
[MINOR] Fix locale specific NumberFormatException in testutils HoodieTestDataGenerator (#7215)
---
.../java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
index 5cedea06342..c6ab11a310f 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
@@ -68,6 +68,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Random;
import java.util.Set;
@@ -363,7 +364,7 @@ public class HoodieTestDataGenerator implements AutoCloseable {
rec.put("current_date", (int) LocalDateTime.ofInstant(instant, ZoneOffset.UTC).toLocalDate().toEpochDay());
rec.put("current_ts", randomMillis);
- BigDecimal bigDecimal = new BigDecimal(String.format("%5f", rand.nextFloat()));
+ BigDecimal bigDecimal = new BigDecimal(String.format(Locale.ENGLISH, "%5f", rand.nextFloat()));
Schema decimalSchema = AVRO_SCHEMA.getField("height").schema();
Conversions.DecimalConversion decimalConversions = new Conversions.DecimalConversion();
GenericFixed genericFixed = decimalConversions.toFixed(bigDecimal, decimalSchema, LogicalTypes.decimal(10, 6));
[hudi] 05/17: [HUDI-5179] Updated Hudi Release guide (#7212)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 1e8315a4ce4fd3960a6f01e3a2f6fcdd0371b02b
Author: Zhaojing Yu <yu...@bytedance.com>
AuthorDate: Tue Dec 6 08:04:54 2022 +0800
[HUDI-5179] Updated Hudi Release guide (#7212)
---
release/release_guide.md | 680 +++++++++++++++++++++++++++++++++++++
release/release_guide_overview.jpg | Bin 0 -> 363329 bytes
2 files changed, 680 insertions(+)
diff --git a/release/release_guide.md b/release/release_guide.md
new file mode 100644
index 00000000000..41a2ea95341
--- /dev/null
+++ b/release/release_guide.md
@@ -0,0 +1,680 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+
+# Introduction
+
+This release process document is based on [Apache Beam Release Guide](https://beam.apache.org/contribute/release-guide/)
+and [Apache Flink Release Guide](https://cwiki.apache.org/confluence/display/FLINK/Creating+a+Flink+Release).
+
+The older version of the release guide [Apache Hudi Release Guide](https://cwiki.apache.org/confluence/display/HUDI/Apache+Hudi+-+Release+Guide).
+
+The Apache Hudi project periodically declares and publishes releases. A release is one or more packages of the project
+artifact(s) that are approved for general public distribution and use. They may come with various degrees of caveat
+regarding their perceived quality and potential for change, such as “alpha”, “beta”, “stable”, etc.
+
+Hudi community treats releases with great importance. They are a public face of the project and most users interact with
+the project only through the releases. Releases are signed off by the entire Hudi community in a public vote.
+
+Each release is executed by a Release Manager, who is selected among the Hudi PMC members. This document describes the
+process that the Release Manager follows to perform a release. Any changes to this process should be discussed and
+adopted on the dev@ mailing list.
+
+Please remember that publishing software has legal consequences. This guide complements the
+foundation-wide [Product Release Policy](http://www.apache.org/dev/release.html)
+and [Release Distribution Policy](http://www.apache.org/dev/release-distribution).
+
+# Overview
+
+
+
+The release process consists of several steps:
+
+1. Decide to release
+2. Prepare for the release
+3. Build a release candidate
+4. Vote on the release candidate
+5. During vote process, run validation tests
+6. If necessary, fix any issues and go back to step 3.
+7. Finalize the release
+8. Promote the release
+
+# Decide to release
+
+Deciding to release and selecting a Release Manager is the first step of the release process. This is a consensus-based
+decision of the entire community.
+
+Anybody can propose a release on the dev@ mailing list, giving a solid argument and nominating a committer as the
+Release Manager (including themselves). There’s no formal process, no vote requirements, and no timing requirements. Any
+objections should be resolved by consensus before starting the release.
+
+In general, the community prefers to have a rotating set of 3-5 Release Managers. Keeping a small core set of managers
+allows enough people to build expertise in this area and improve processes over time, without Release Managers needing
+to re-learn the processes for each release. That said, if you are a committer interested in serving the community in
+this way, please reach out to the community on the dev@ mailing list.
+
+## Checklist to proceed to the next step
+
+1. Community agrees to release
+2. Community selects a Release Manager
+
+# Prepare for the release
+
+As a release manager, you should create a private Slack channel, named `hudi-<version>_release_work` (e.g.
+hudi-0_12_0_release_work) in Apache Hudi Slack for coordination. Invite all committers to the channel.
+
+Before your first release, you should perform one-time configuration steps. This will set up your security keys for
+signing the release and access to various release repositories.
+
+To prepare for each release, you should audit the project status in the JIRA issue tracker, and do the necessary
+bookkeeping. Finally, you should create a release branch from which individual release candidates will be built.
+
+**NOTE**: If you are
+using [GitHub two-factor authentication](https://help.github.com/articles/securing-your-account-with-two-factor-authentication-2fa/)
+and haven’t configure HTTPS access, please
+follow [the guide](https://help.github.com/articles/creating-a-personal-access-token-for-the-command-line/) to configure
+command line access.
+
+## One-time Setup Instructions
+
+You need to have a GPG key to sign the release artifacts. Please be aware of the
+ASF-wide [release signing guidelines](https://www.apache.org/dev/release-signing.html). If you don’t have a GPG key
+associated with your Apache account, please follow the section below.
+
+### For Linux users
+
+There are 2 ways to configure your GPG key for release, either using release automation script(which is recommended), or
+running all commands manually. If using Mac, please see below to handle known issues.
+
+#### Use preparation_before_release.sh to setup GPG
+
+- Script: preparation_before_release.sh
+- Usage ./hudi/scripts/release/preparation_before_release.sh
+- Tasks included
+ 1. Help you create a new GPG key if you want.
+ 2. Configure git user.signingkey with chosen pubkey.
+ 3. Add chosen pubkey into dev KEYS and release KEYS **NOTES**: Only PMC can write into release repo.
+ 4. Start GPG agents.
+
+#### Run all commands manually
+
+- Get more entropy for creating a GPG key
+ - sudo apt-get install rng-tools
+ - sudo rngd -r /dev/urandom
+- Create a GPG key
+ - gpg --full-generate-key
+- Determine your Apache GPG Key and Key ID, as follows:
+ - gpg --list-keys
+- This will list your GPG keys. One of these should reflect your Apache account, for example:
+ ```
+ --------------------------------------------------
+ pub 2048R/935D191 2019-08-29
+ uid Anonymous Anonymous <an...@apache.org>
+ sub 2048R/CD4C59FD 2019-08-29
+ ```
+ Here, the key ID is the 8-digit hex string in the pub line: 845E6689 or more than 8-digit hex string like
+ 623E08E06DB376684FB9599A3F5953147903948A. Now, add your Apache GPG key to the Hudi’s KEYS file both
+ in [dev](https://dist.apache.org/repos/dist/dev/hudi/KEYS)
+ and [release](https://dist.apache.org/repos/dist/release/hudi/KEYS) repositories
+ at [dist.apache.org](http://dist.apache.org/). Follow the instructions listed at the top of these files. (**Note**:
+ Only PMC members have write access to the release repository. If you end up getting 403 errors ask on the mailing list
+ for assistance.)
+- Configure git to use this key when signing code by giving it your key ID, as follows:
+ - git config --global user.signingkey XXXXXXXX
+ - You may drop the --global option if you’d prefer to use this key for the current repository only.
+- Start GPG agent in order to unlock your GPG key
+ - eval $(gpg-agent --daemon --no-grab --write-env-file $HOME/.gpg-agent-info)
+ - export GPG_TTY=$(tty)
+ - export GPG_AGENT_INFO
+
+### For Mac users
+
+- apt-get is not available. So install gpg using [https://gpgtools.org/](https://gpgtools.org/).
+- Create gpg key with your apache emailId and publish to key server (referhttps://gpgtools.org/ to the section "Submit
+ your GPG public key into MIT PGP Public Key Server" below)
+- The KEYS file is in [https://dist.apache.org/repos/dist/](https://dist.apache.org/repos/dist/)
+ - To checkout you need subversion. If subversion is not available in Mac you might have to first install it
+ using `brew install subversion`.
+
+### Access to Apache Nexus repository
+
+Configure access to the Apache Nexus repository, which enables final deployment of releases to the Maven Central
+Repository.
+
+1. You log in with your Apache account.
+2. Confirm you have appropriate access by finding org.apache.hudi under Staging Profiles.
+3. Navigate to your Profile (top right dropdown menu of the page).
+4. Choose User Token from the dropdown, then click Access User Token. Copy a snippet of the Maven XML configuration
+ block.
+5. Insert this snippet twice into your global Maven settings.xml file, typically ${HOME}/.m2/settings.xml. The end
+ result should look like this, where TOKEN_NAME and TOKEN_PASSWORD are your secret tokens:
+
+```xml
+<settings>
+ <servers>
+ <server>
+ <id>apache.releases.https</id>
+ <username>TOKEN_NAME</username>
+ <password>TOKEN_PASSWORD</password>
+ </server>
+ <server>
+ <id>apache.snapshots.https</id>
+ <username>TOKEN_NAME</username>
+ <password>TOKEN_PASSWORD</password>
+ </server>
+ </servers>
+ </settings>
+```
+
+### Submit your GPG public key into MIT PGP Public Key Server
+
+In order to make yourself have the right permission to stage java artifacts in Apache Nexus staging repository, please
+submit your GPG public key into [MIT PGP Public Key Server](http://pgp.mit.edu:11371/).
+
+Also send public key to ubuntu server via
+
+```shell
+gpg --keyserver hkp://keyserver.ubuntu.com --send-keys ${PUBLIC_KEY} # send public key to ubuntu server
+gpg --keyserver hkp://keyserver.ubuntu.com --recv-keys ${PUBLIC_KEY} # verify
+```
+
+would also refer
+to [stackoverflow](https://stackoverflow.com/questions/19462617/no-public-key-key-with-id-xxxxx-was-not-able-to-be-located-oss-sonatype-org)
+.
+
+## Create a new version in JIRA
+
+When contributors resolve an issue in JIRA, they are tagging it with a release that will contain their changes. With the
+release currently underway, new issues should be resolved against a subsequent future release. Therefore, you should
+create a release item for this subsequent release, as follows:
+
+**Attention**: Only PMC has permission to perform this. If you are not a PMC, please ask for help in dev@ mailing list.
+
+In [JIRA](https://jira.apache.org/jira/projects/HUDI/summary), navigate
+to [Hudi > Administration > Versions](https://issues.apache.org/jira/plugins/servlet/project-config/HUDI/versions). Add
+a new release. Choose the next minor version number after the version currently underway, select the release cut date (
+today’s date) as the Start Date, and choose Add. At the end of the release, go to the same page and mark the recently
+released version as released. Use the ... menu and choose Release.
+
+## Triage release-blocking issues in JIRA
+
+There could be outstanding release-blocking issues, which should be triaged before proceeding to build a release
+candidate. We track them by assigning a specific Fix version field even before the issue resolved.
+
+The list of release-blocking issues is available at
+the [version status page](https://issues.apache.org/jira/projects/HUDI?selectedItem=com.atlassian.jira.jira-projects-plugin%3Arelease-page&status=unreleased-archived).
+Triage each unresolved issue with one of the following resolutions:
+
+For all JIRA issues:
+- If the issue has been resolved and JIRA was not updated, resolve it accordingly.
+
+For JIRA issues with type “Bug” or labeled “flaky”:
+- If the issue is a known continuously failing test, it is not acceptable to defer this until the next release. Please
+work with the Hudi community to resolve the issue.
+- If the issue is a known flaky test, make an attempt to delegate a fix. However, if the issue may take too long to fix (to the discretion of the release manager):
+ - Delegate manual testing of the flaky issue to ensure no release blocking issues.
+ - Update the Fix Version field to the version of the next release. Please consider discussing this with stakeholders and the dev@ mailing list, as appropriate.
+
+For all other JIRA issues:
+- If the issue has not been resolved and it is acceptable to defer this until the next release, update the Fix Version field to the new version you just created. Please consider discussing this with stakeholders and the dev@ mailing list, as appropriate.
+- If the issue has not been resolved and it is not acceptable to release until it is fixed, the release cannot proceed. Instead, work with the Hudi community to resolve the issue.
+
+If there is a bug found in the RC creation process/tools, those issues should be considered high priority and fixed in 7 days.
+
+## Review Release Notes in JIRA
+
+JIRA automatically generates Release Notes based on the Fix Version field applied to issues. Release Notes are intended for Hudi users (not Hudi committers/contributors). You should ensure that Release Notes are informative and useful.
+
+Open the release notes from the [version status page](https://issues.apache.org/jira/browse/Hudi/?selectedTab=com.atlassian.jira.jira-projects-plugin:versions-panel) by choosing the release underway and clicking Release Notes.
+
+You should verify that the issues listed automatically by JIRA are appropriate to appear in the Release Notes. Specifically, issues should:
+
+- Be appropriately classified as Bug, New Feature, Improvement, etc.
+- Represent noteworthy user-facing changes, such as new functionality, backward-incompatible API changes, or performance improvements.
+- Have occurred since the previous release; an issue that was introduced and fixed between releases should not appear in the Release Notes.
+- Have an issue title that makes sense when read on its own.
+
+## Create a release branch in apache/hudi repository
+
+Attention: Only committer has permission to create release branch in apache/hudi.
+**Skip this step if it is a bug fix release. But do set the env variables.**
+
+Release candidates are built from a release branch. As a final step in preparation for the release, you should create the release branch, push it to the Apache code repository, and update version information on the original branch.
+
+Export Some Environment variables in the terminal where you are running the release scripts
+
+- export RELEASE_VERSION=<RELEASE_VERSION_TO_BE_PUBLISHED>
+- export NEXT_VERSION=<NEW_VERSION_IN_MASTER>
+- export RELEASE_BRANCH=release-<RELEASE_VERSION_TO_BE_PUBLISHED>
+- export RC_NUM=<release_candidate_num_starting_from_1>
+
+Use cut_release_branch.sh to cut a release branch
+
+- Script: [cut_release_branch.sh](https://github.com/apache/incubator-hudi/blob/master/scripts/release/cut_release_branch.sh)
+
+Usage
+
+```shell
+# Cut a release branch
+Cd scripts && ./release/cut_release_branch.sh \
+--release=${RELEASE_VERSION} \
+--next_release=${NEXT_VERSION} \
+--rc_num=${RC_NUM}
+# Show help page
+./hudi/scripts/release/cut_release_branch.sh -h
+```
+
+## For Bug Fix release:
+
+Here is how to go about a bug fix release.
+
+- Create a branch in your repo (<user>/hudi).
+- Cherry-pick commits from master that needs to be part of this release. (git cherry-pick commit-hash). You need to manually resolve the conflicts. For eg, a file might have been moved to a diff class in master where as in your release branch, it could be in older place. You need to take a call where to place it. Similar things like file addition, file deletion, etc.
+- Update the release version by running "mvn versions:set -DnewVersion=${RELEASE}-rc${RC_NUM}", with "RELEASE" as the version and "RC_NUM" as the RC number. Make sure the version changes are intended. Then git commit the changes.
+- Ensure both compilation and tests are good.
+- I assume you will have apache/hudi as upstream. If not add it as upstream
+- Once the branch is ready with all commits, go ahead and push your branch to upstream.
+- Go to apache/hudi repo locally and pull this branch. Here after you can work on this branch and push to origin when need be.
+- Do not forget to set the env variables from above section.
+
+## Verify that a Release Build Works
+
+Run "mvn -Prelease clean install" to ensure that the build processes are in good shape. // You need to execute this command once you have the release branch in apache/hudi
+
+Good to run this for all profiles
+
+```shell
+mvn -Prelease clean install
+mvn -Prelease clean install -Dscala-2.12
+mvn -Prelease clean install -Dspark3
+```
+
+## Checklist to proceed to the next step
+
+- Release Manager’s GPG key is published to [dist.apache.org](http://dist.apache.org/).
+- Release Manager’s GPG key is configured in git configuration.
+- Release Manager has org.apache.hudi listed under Staging Profiles in Nexus.
+- Release Manager’s Nexus User Token is configured in settings.xml.
+- JIRA release item for the subsequent release has been created.
+- All test failures from branch verification have associated JIRA issues.
+- There are no release blocking JIRA issues.
+- Release branch has been created.
+- Release Notes have been audited and added to RELEASE_NOTES.md.
+
+# Build a release candidate
+
+The core of the release process is the build-vote-fix cycle. Each cycle produces one release candidate. The Release Manager repeats this cycle until the community approves one release candidate, which is then finalized.
+
+Set up a few environment variables to simplify Maven commands that follow. This identifies the release candidate being built. Start with RC_NUM equal to 1 and increment it for each candidate.
+
+1. git checkout ${RELEASE_BRANCH}
+2. Run mvn version to set the proper rc number in all artifacts
+ 1. mvn versions:set -DnewVersion=${RELEASE_VERSION}-rc${RC_NUM}
+3. Run Unit tests and ensure they succeed
+ 1. mvn test -DskipITs=true
+4. Run Integration Tests and ensure they succeed
+ 1. mvn verify -DskipUTs=true
+5. Commit and push this change to RELEASE branch
+ 1. git commit -am "Bumping release candidate number ${RC_NUM}"
+ > There will be some backup files created which needs to be removed.
+ >
+ > You could do "git clean -fd" before doing the commit.
+ 2. git push origin ${RELEASE_BRANCH}
+ > If you already have a remote tag with same name as your branch, you can try below command.
+ >
+ > git push origin refs/heads/${RELEASE_BRANCH}
+ >
+ > "refs/heads/" refers to a branch.
+ >
+ > "refs/tags/" refers to tag.
+6. Generate Source Release: This will create the tarball under hudi/src_release directory
+ 1. git checkout ${RELEASE_BRANCH}
+ 2. cd scripts && ./release/create_source_release.sh
+ > If you have multiple gpg keys(gpg --list-keys), then the signing command will take in the first key most likely.
+ > You will release this when it asks for a passphrase in a pop up.
+ > When asked for passphrase, ensure the intended key is the one asked for.
+ >
+ > Command used in script:
+ >
+ > gpg --armor --detach-sig ${RELEASE_DIR}/hudi-${RELEASE_VERSION}.src.tgz
+ >
+ > To use a specific key: update as follows: // replace FINGERPRINT
+ >
+ > gpg --local-user [FINGERPRINT] --armor --detach-sig ${RELEASE_DIR}/hudi-${RELEASE_VERSION}.src.tgz
+ 3. Verify Source release is signed and buildable
+ 1. cd hudi/src_release
+ 2. gpg --verify hudi-${RELEASE_VERSION}-rc${RC_NUM}.src.tgz.asc hudi-${RELEASE_VERSION}-rc${RC_NUM}.src.tgz
+ 3. tar -zxvf hudi-${RELEASE_VERSION}-rc${RC_NUM}.src.tgz && cd hudi-${RELEASE_VERSION}-rc${RC_NUM} && mvn clean package -DskipTests -Pintegration-tests
+ 4. If they pass, delete the repository we got from the tar-ball
+ - cd ../ && rm -rf hudi-${RELEASE_VERSION}-rc${RC_NUM}
+
+7. Create tag
+ 1. git tag -s release-${RELEASE_VERSION}-rc${RC_NUM} -m "${RELEASE_VERSION}".
+ > If you run into some issues, and if want to re-run the same candidate again from start, ensure you delete existing tags before retrying again.
+ >
+ > // to remove local
+ >
+ > git tag -d release-${RELEASE_VERSION}-rc${RC_NUM}
+ >
+ > // to remove remote
+ >
+ > git push --delete origin release-${RELEASE_VERSION}-rc${RC_NUM}
+ 2. if apache repo is origin.
+ > git push origin release-${RELEASE_VERSION}-rc${RC_NUM}
+ >
+ > If a branch with the same name already exists in origin, this command might fail as below.
+ >
+ > error: src refspec release-0.5.3 matches more than one
+ >
+ > error: failed to push some refs to 'https://github.com/apache/hudi.git'
+ >
+ > In such a case, try below command
+ >
+ > git push origin refs/tags/release-${RELEASE_VERSION}-rc${RC_NUM}
+
+8. [Stage source releases](https://www.apache.org/legal/release-policy.html#stage) on [dist.apache.org](http://dist.apache.org/)
+ 1. If you have not already, check out the Hudi section of the dev repository on [dist.apache.org](http://dist.apache.org/) via Subversion. In a fresh directory
+ 2. if you would not checkout, please try svn checkout [https://dist.apache.org/repos/dist/dev/hudi](https://dist.apache.org/repos/dist/dev/hudi) again.
+ 1. svn checkout [https://dist.apache.org/repos/dist/dev/hudi](https://dist.apache.org/repos/dist/dev/hudi) --depth=immediates
+ 3. Make a directory for the new release:
+ 1. mkdir hudi/hudi-${RELEASE_VERSION}-rc${RC_NUM}
+ 4. Copy Hudi source distributions, hashes, and GPG signature:
+ 1. mv <hudi-dir>/src_release/* hudi/hudi-${RELEASE_VERSION}-rc${RC_NUM}
+ 5. Add and commit all the files.
+ 1. cd hudi
+ 2. svn add hudi-${RELEASE_VERSION}-rc${RC_NUM}
+ 3. svn commit
+ 6. Verify that files are [present](https://dist.apache.org/repos/dist/dev/hudi)
+ 7. Run Verification Script to ensure the source release is sane
+ 1. For RC: cd scripts && ./release/validate_staged_release.sh --release=${RELEASE_VERSION} --rc_num=${RC_NUM} --verbose
+ 2. For finalized release in dev: cd scripts && ./release/validate_staged_release.sh --release=${RELEASE_VERSION} --verbose
+
+9. Deploy maven artifacts and verify
+ 1. This will deploy jar artifacts to the Apache Nexus Repository, which is the staging area for deploying jars to Maven Central.
+ 2. Review all staged artifacts (https://repository.apache.org/). They should contain all relevant parts for each module, including pom.xml, jar, test jar, source, test source, javadoc, etc. Carefully review any new artifacts.
+ 3. git checkout ${RELEASE_BRANCH}
+ 4. ./scripts/release/deploy_staging_jars.sh 2>&1 | tee -a "/tmp/${RELEASE_VERSION}-${RC_NUM}.deploy.log"
+ 1. when prompted for the passphrase, if you have multiple gpg keys in your keyring, make sure that you enter the right passphase corresponding to the same key (FINGERPRINT) as used while generating source release in step f.ii.
+ > If the prompt is not for the same key (by default the maven-gpg-plugin will pick up the first key in your keyring so that could be different), then add the following option to your ~/.gnupg/gpg.conf file
+ 2. make sure your IP is not changing while uploading, otherwise it creates a different staging repo
+ 3. Use a VPN if you can't prevent your IP from switching
+ 4. after uploading, inspect the log to make sure all maven tasks said "BUILD SUCCESS"
+ 5. Review all staged artifacts by logging into Apache Nexus and clicking on "Staging Repositories" link on left pane. Then find a "open" entry for apachehudi
+ 6. Ensure it contains all 3 (2.11, 2.12 with spark2 and 2.12 with spark3) artifacts, mainly hudi-spark-bundle-2.11/2.12, hudi-spark3-bundle-2.12, hudi-spark-2.11/2.12, hudi-spark2-2.11/2.12, hudi-spark3-2.12, hudi-utilities-bundle_2.11/2.12 and hudi-utilities_2.11/2.12.
+ > With 0.10.1, we had 4 bundles. spark2 with scala11, spark2 with scala12, spark3.0.x bundles and spark3.1.x bundles. Ensure each spark bundle reflects the version correctly. hudi-spark3.1.2-bundle_2.12-0.10.1.jar and hudi-spark3.0.3-bundle_2.12-0.10.1.jar are the respective bundle names for spark3 bundles.
+ 7. Once you have ensured everything is good and validation of step 7 succeeds, you can close the staging repo. Until you close, you can re-run deploying to staging multiple times. But once closed, it will create a new staging repo. So ensure you close this, so that the next RC (if need be) is on a new repo. So, once everything is good, close the staging repository on Apache Nexus. When prompted for a description, enter
+ > Apache Hudi, version `${RELEASE_VERSION}`, release candidate `${RC_NUM}`.
+ 8. After closing, run the script to validate the staged bundles again:
+ ```shell
+ ./scripts/release/validate_staged_bundles.sh orgapachehudi-<stage_repo_number> ${RELEASE_VERSION}-rc${RC_NUM} 2>&1 | tee -a /tmp/validate_staged_bundles_output.txt
+ ```
+
+## Checklist to proceed to the next step
+
+1. Maven artifacts deployed to the staging repository of repository.apache.org
+2. Source distribution deployed to the dev repository of dist.apache.org and validated
+
+# Vote on the release candidate
+
+Once you have built and individually reviewed the release candidate, please share it for the community-wide(dev@hudi ) review. Please review foundation-wide voting guidelines for more information.
+
+Start the review-and-vote thread on the dev@ mailing list. Here’s an email template; please adjust as you see fit.
+
+```
+From: Release Manager
+
+To: dev@hudi.apache.org
+
+Subject: [VOTE] Release 1.2.3, release candidate #3
+
+
+
+Hi everyone,
+
+Please review and vote on the release candidate #3 for the version 1.2.3, as follows:
+
+[ ] +1, Approve the release
+
+[ ] -1, Do not approve the release (please provide specific comments)
+
+
+
+The complete staging area is available for your review, which includes:
+
+* JIRA release notes [1],
+
+* the official Apache source release and binary convenience releases to be deployed to dist.apache.org [2], which are signed with the key with fingerprint FFFFFFFF [3],
+
+* all artifacts to be deployed to the Maven Central Repository [4],
+
+* source code tag "1.2.3-rc3" [5],
+
+
+
+The vote will be open for at least 72 hours. It is adopted by majority approval, with at least 3 PMC affirmative votes.
+
+
+
+Thanks,
+
+Release Manager
+
+
+
+[1] link
+
+[2] link
+
+[3] https://dist.apache.org/repos/dist/release/hudi/KEYS
+
+[4] link
+
+[5] link
+
+[6] link
+```
+
+If there are any issues found in the release candidate, reply on the vote thread to cancel the vote, and there’s no need to wait 72 hours if any issues found. Proceed to the Fix Issues step below and address the problem. However, some issues don’t require cancellation. For example, if an issue is found in the website, just correct it on the spot and the vote can continue as-is.
+
+If there are no issues, reply on the vote thread to close the voting. Then, tally the votes in a separate email. Here’s an email template; please adjust as you see fit.
+
+```
+From: Release Manager
+
+To: dev@hudi.apache.org
+
+Subject: [RESULT] [VOTE] Release 1.2.3, release candidate #3
+
+
+
+I'm happy to announce that we have unanimously approved this release.
+
+
+
+There are XXX approving votes, XXX of which are binding:
+
+* approver 1
+
+* approver 2
+
+* approver 3
+
+* approver 4
+
+
+
+There are no disapproving votes.
+
+
+
+Thanks everyone!
+```
+
+Please look at previous examples in previous releases. For example : Please see examples here : [voting in dev](https://lists.apache.org/thread.html/617cad3116308821f845d3be5d71e135fdcbb82b4533c53941a0fda8@%3Cdev.hudi.apache.org%3E), [voting in general](https://lists.apache.org/thread.html/14a4c83df04f5bde799dfc904e6fdc67db225ff4e91d945f2e29dae9@%3Cgeneral.incubator.apache.org%3E) and [result of voting](https://lists.apache.org/thread.html/19bbef3a0a5d660b3bd5791346c6b30855443b7b0069e26a [...]
+
+## Checklist to proceed to the finalization step
+
+Community votes to release the proposed candidate, with at least three approving PMC votes
+
+# Fix any issues
+
+Any issues identified during the community review and vote should be fixed in this step.
+
+Code changes should be proposed as standard pull requests to the master branch and reviewed using the normal contributing process. Then, relevant changes should be cherry-picked into the release branch. The cherry-pick commits should then be proposed as the pull requests against the release branch, again reviewed and merged using the normal contributing process.
+
+Once all issues have been resolved, you should go back and build a new release candidate with these changes.
+
+## Checklist to proceed to the next step
+
+Issues identified during vote have been resolved, with fixes committed to the release branch.
+
+# Finalize the release
+
+Once the release candidate has been reviewed and approved by the community, the release should be finalized. This involves the final deployment of the release candidate to the release repositories, merging of the website changes, etc.
+
+## Current follow the below steps to finalize the release.
+
+1. Drop all RC orgapachehudi-XXX in [Apache Nexus Staging Repositories](https://repository.apache.org/#stagingRepositories).
+2. change the version from ${RELEASE_VERSION}-rc${RC_NUM} to ${RELEASE_VERSION} against release branch, use command `mvn versions:set -DnewVersion=${RELEASE_VERSION}`, e.g. change 0.5.1-rc1 to 0.5.1.
+3. Commit and push the version change to release branch.
+ 1. git commit -am "[MINOR] Update release version to reflect published version ${RELEASE_VERSION}"
+ 2. git push origin release-${RELEASE_VERSION}
+4. Repeat the steps from **_Generate Source Release (f) to Stage source releases on [dist.apache.org](http://dist.apache.org/) (i)_**. Including staging jars with the release version and uploading source release.
+ > **Note that make sure remove the -rc${RC_NUM} suffix when repeat the above steps. and please also verify the steps. Ensure git tag is also done without -rc${RC_NUM}**
+5. One more step is to [deploy source code to release](https://www.apache.org/legal/release-policy.html#upload-ci) dist. [https://dist.apache.org/repos/dist/release/hudi](https://dist.apache.org/repos/dist/release/hudi).
+ Only PMC will have access to this repo. So, if you are not a PMC, do get help from somone who is.
+ 1. svn checkout https://dist.apache.org/repos/dist/release/hudi --depth=immediates, if you would not checkout, please try svn checkout https://dist.apache.org/repos/dist/release/hudi again.
+ 2. Make a directory for the new release:
+ ```shell
+ mkdir hudi/${RELEASE_VERSION}
+ ```
+ 3. Copy Hudi source distributions, hashes, and GPG signature:
+ ```shell
+ mv <hudi-dir>/src_release/* hudi/${RELEASE_VERSION}
+ ```
+ 4. Add and commit all the files.
+ ```shell
+ cd hudi
+ svn add ${RELEASE_VERSION}
+ svn commit
+ ```
+ 5. Verify that files are [present](https://dist.apache.org/repos/dist/release/hudi)
+6. Use the Apache Nexus repository to release the staged binary artifacts to the Maven Central repository. In the Staging Repositories section, find the relevant release candidate orgapachehudi-XXX entry and click `Release`.
+ > Note: make sure and check you click `Release` repo, it cannot be withdrawn.
+7. Drop all other release candidates that are not being released. It can take up to 24 hours for the new release to show up in [Maven Central repository](https://search.maven.org/search?q=g:org.apache.hudi).
+8. In Jira, go to Releases → <Release Version> and ensure that all Jiras for the release are 'Closed' state, if not transition all 'Resolved' jiras to 'Closed'.
+ > Note: we should close jira and choose 'resolution = Fixed' rather than resolve jira.
+9. Finalize the Release in Jira by providing the release date.
+10. Update [DOAP](https://github.com/apache/hudi/blob/master/doap_HUDI.rdf) file in the root of the project via sending a PR like [this one](https://github.com/apache/incubator-hudi/pull/1448).
+11. Create a new Github release, off the release version tag, you pushed before
+
+## Steps to cut doc version and update website.
+
+1. Follow the [instructions](https://github.com/apache/hudi/blob/asf-site/README.md#adding-docs-for-version) for cutting a doc for this new release.
+ - For a minor release, make sure any docs updates regarding the new features for the next major release, not in the minor release, are excluded. This involves reverting the changes or manually removing the relevant docs in a separate branch.
+ - Adding examples of commits from 0.12.1 that you can follow:
+ 1. https://github.com/apache/hudi/commit/1677b2a7b3a863e96022ce0d8c658694a9ad6686
+ 2. https://github.com/apache/hudi/commit/6645380ef21508fb6fc646aa42692fa9e3a53f6c
+2. [Build the site locally](https://github.com/apache/hudi/blob/asf-site/README.md#building-docs) and ensure the new doc version is available as intended.
+3. Update site using [instructions](https://github.com/apache/hudi/blob/asf-site/README.md#updating-site)
+4. There are few adhoc fixes that needs to be taken care apart from above steps. Adding examples of commits from 0.10.1 that you can follow.
+ 1. https://github.com/apache/hudi/pull/4703/commits/1e7c9af976b075c9ed5a780da1cdb57f019a41d3
+ 2. https://github.com/apache/hudi/pull/4703/commits/b474ec266fe2243f8146ba7a112045bbc8b0ddc8
+ 3. https://github.com/apache/hudi/commit/549fa7a51b30b162dcd6fc70b42cf1779de1900b
+ 4. https://github.com/apache/hudi/commit/de3405855c23aeb449113de197591c186396a4c2
+
+## Released version in jira releases
+
+1. Make sure all issues in the release are closed.
+2. Go to the same page and mark the recently released version as released. Use the ... menu and choose Release.`.
+
+# Promote the release
+
+Once the release has been finalized, the last step of the process is to promote the release within the project and beyond. Please wait for 1h after finalizing the release in accordance with the [ASF release policy](http://www.apache.org/legal/release-policy.html#release-announcements).
+
+## Apache mailing lists
+
+Announce on the dev@ mailing list that the release has been finished.
+
+Announce on the release on the user@ mailing list, listing major improvements and contributions.
+
+Announce the release on the [announce@apache.org](announce@apache.org) mailing list. **NOTE**: put [announce@apache.org](announce@apache.org) in **bcc** to avoid disrupting people with followups.
+
+Considering that announce@ ML has restrictions on what is published, we can follow this email template:
+
+```
+From: Release Manager
+
+To: announce@apache.org
+
+Subject: [ANNOUNCE] Apache Hudi <VERSION> released
+
+
+
+The Apache Hudi team is pleased to announce the release of Apache
+
+Hudi <VERSION>.
+
+
+
+Apache Hudi (pronounced Hoodie) stands for Hadoop Upserts Deletes
+
+and Incrementals. Apache Hudi manages storage of large analytical
+
+datasets on DFS (Cloud stores, HDFS or any Hadoop FileSystem compatible storage)
+
+and provides the ability to query them.
+
+
+
+This release comes xxx months after xxx. It includes more than
+xxx resolved issues, comprising of a few new features as well as
+general improvements and bug-fixes. It includes support for
+xxx, xxx, xxx, and many more bug fixes and improvements.
+
+For details on how to use Hudi, please look at the quick start page located at https://hudi.apache.org/docs/quick-start-guide.html
+
+If you'd like to download the source release, you can find it here:
+
+https://github.com/apache/hudi/releases/tag/<VERSION>
+
+You can read more about the release (including release notes) here:
+
+https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12322822&version=<JIRA_VERSION>
+
+We welcome your help and feedback. For more information on how to
+report problems, and to get involved, visit the project website at:
+
+https://hudi.apache.org/
+
+Thanks to everyone involved!
+
+XXX
+```
+
+## Recordkeeping
+
+Use [reporter.apache.org](http://reporter.apache.org/) to seed the information about the release into future project reports.
+
+## Social media
+
+Tweet, post on Facebook, LinkedIn, and other platforms. Ask other contributors to do the same.
+
+# Improve the process
+
+It is important that we improve the release processes over time. Once you’ve finished the release, please take a step back and look what areas of this process and be improved. Perhaps some part of the process can be simplified. Perhaps parts of this guide can be clarified.
+
+If we have specific ideas, please start a discussion on the dev@ mailing list and/or propose a pull request to update this guide. Thanks!
\ No newline at end of file
diff --git a/release/release_guide_overview.jpg b/release/release_guide_overview.jpg
new file mode 100644
index 00000000000..bf7602ddeee
Binary files /dev/null and b/release/release_guide_overview.jpg differ
[hudi] 13/17: [HUDI-5344] Fix CVE - upgrade protobuf-java to 3.18.2 (#6957)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit aef045b8ce7876dae8b1711253062ab00bc6c476
Author: taorong <14...@users.noreply.github.com>
AuthorDate: Wed Dec 7 20:29:14 2022 +0800
[HUDI-5344] Fix CVE - upgrade protobuf-java to 3.18.2 (#6957)
Upgrade necessary to include fix for
https://www.oscs1024.com/hd/CVE-2021-22569
---
packaging/hudi-presto-bundle/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/packaging/hudi-presto-bundle/pom.xml b/packaging/hudi-presto-bundle/pom.xml
index aa2f58c8eff..73d77ab6bb7 100644
--- a/packaging/hudi-presto-bundle/pom.xml
+++ b/packaging/hudi-presto-bundle/pom.xml
@@ -300,7 +300,7 @@
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
- <version>3.17.3</version>
+ <version>3.18.2</version>
<scope>${presto.bundle.bootstrap.scope}</scope>
</dependency>
[hudi] 01/17: [HUDI-5306] Unify RecordIterator and HoodieParquetReader with ClosableIterator (#7340)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d228a10770ac97f9f3407626839da27470ab89da
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Thu Dec 1 17:13:59 2022 +0800
[HUDI-5306] Unify RecordIterator and HoodieParquetReader with ClosableIterator (#7340)
* Unify RecordIterator and HoodieParquetReader with ClosableIterator
* Add a factory clazz for RecordIterator
* Add more documents
---
.../apache/hudi/configuration/OptionsResolver.java | 2 +-
.../java/org/apache/hudi/table/format/CastMap.java | 19 +-
...Reader.java => ParquetSplitRecordIterator.java} | 24 ++-
...odieParquetReader.java => RecordIterators.java} | 56 +++---
...eader.java => SchemaEvolvedRecordIterator.java} | 25 ++-
.../table/format/cow/CopyOnWriteInputFormat.java | 19 +-
.../table/format/mor/MergeOnReadInputFormat.java | 195 +++++++++------------
7 files changed, 159 insertions(+), 181 deletions(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
index dd272e17fb0..2d56ea4e043 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
@@ -207,7 +207,7 @@ public class OptionsResolver {
* Returns whether comprehensive schema evolution enabled.
*/
public static boolean isSchemaEvolutionEnabled(Configuration conf) {
- return conf.getBoolean(HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key(), false);
+ return conf.getBoolean(HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key(), HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.defaultValue());
}
// -------------------------------------------------------------------------
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/CastMap.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/CastMap.java
index 5f29e85adc2..36cf8708875 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/CastMap.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/CastMap.java
@@ -49,14 +49,17 @@ import static org.apache.flink.table.types.logical.LogicalTypeRoot.VARCHAR;
/**
* CastMap is responsible for conversion of flink types when full schema evolution enabled.
- * Supported cast conversions:
- * Integer => Long, Float, Double, Decimal, String
- * Long => Float, Double, Decimal, String
- * Float => Double, Decimal, String
- * Double => Decimal, String
- * Decimal => Decimal, String
- * String => Decimal, Date
- * Date => String
+ *
+ * <p>Supported cast conversions:
+ * <ul>
+ * <li>Integer => Long, Float, Double, Decimal, String</li>
+ * <li>Long => Float, Double, Decimal, String</li>
+ * <li>Float => Double, Decimal, String</li>
+ * <li>Double => Decimal, String</li>
+ * <li>Decimal => Decimal, String</li>
+ * <li>String => Decimal, Date</li>
+ * <li>Date => String</li>
+ * </ul>
*/
public final class CastMap implements Serializable {
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetSplitReader.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/ParquetSplitRecordIterator.java
similarity index 63%
rename from hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetSplitReader.java
rename to hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/ParquetSplitRecordIterator.java
index d13c6c7c21a..7b26d71f115 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/ParquetSplitRecordIterator.java
@@ -18,6 +18,8 @@
package org.apache.hudi.table.format;
+import org.apache.hudi.common.util.ClosableIterator;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
import org.apache.flink.table.data.RowData;
@@ -27,25 +29,33 @@ import java.io.IOException;
/**
* Hoodie wrapper for flink parquet reader.
*/
-public final class HoodieParquetSplitReader implements HoodieParquetReader {
+public final class ParquetSplitRecordIterator implements ClosableIterator<RowData> {
private final ParquetColumnarRowSplitReader reader;
- public HoodieParquetSplitReader(ParquetColumnarRowSplitReader reader) {
+ public ParquetSplitRecordIterator(ParquetColumnarRowSplitReader reader) {
this.reader = reader;
}
@Override
- public boolean reachedEnd() throws IOException {
- return reader.reachedEnd();
+ public boolean hasNext() {
+ try {
+ return !reader.reachedEnd();
+ } catch (IOException e) {
+ throw new HoodieIOException("Decides whether the parquet columnar row split reader reached end exception", e);
+ }
}
@Override
- public RowData nextRecord() {
+ public RowData next() {
return reader.nextRecord();
}
@Override
- public void close() throws IOException {
- reader.close();
+ public void close() {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new HoodieIOException("Close the parquet columnar row split reader exception", e);
+ }
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetReader.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/RecordIterators.java
similarity index 67%
rename from hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetReader.java
rename to hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/RecordIterators.java
index e762f03e983..8657f16ddc9 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetReader.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/RecordIterators.java
@@ -18,6 +18,7 @@
package org.apache.hudi.table.format;
+import org.apache.hudi.common.util.ClosableIterator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.table.format.cow.ParquetSplitReaderUtil;
@@ -26,23 +27,17 @@ import org.apache.hudi.util.RowDataProjection;
import org.apache.flink.core.fs.Path;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
-
import org.apache.hadoop.conf.Configuration;
-import java.io.Closeable;
import java.io.IOException;
import java.util.Map;
/**
- * Base interface for hoodie parquet readers.
+ * Factory clazz for record iterators.
*/
-public interface HoodieParquetReader extends Closeable {
-
- boolean reachedEnd() throws IOException;
+public abstract class RecordIterators {
- RowData nextRecord();
-
- static HoodieParquetReader getReader(
+ public static ClosableIterator<RowData> getParquetRecordIterator(
InternalSchemaManager internalSchemaManager,
boolean utcTimestamp,
boolean caseSensitive,
@@ -55,10 +50,9 @@ public interface HoodieParquetReader extends Closeable {
Path path,
long splitStart,
long splitLength) throws IOException {
- Option<RowDataProjection> castProjection;
InternalSchema fileSchema = internalSchemaManager.getFileSchema(path.getName());
if (fileSchema.isEmptySchema()) {
- return new HoodieParquetSplitReader(
+ return new ParquetSplitRecordIterator(
ParquetSplitReaderUtil.genPartColumnarRowReader(
utcTimestamp,
caseSensitive,
@@ -73,27 +67,25 @@ public interface HoodieParquetReader extends Closeable {
splitLength));
} else {
CastMap castMap = internalSchemaManager.getCastMap(fileSchema, fieldNames, fieldTypes, selectedFields);
- castProjection = castMap.toRowDataProjection(selectedFields);
- fieldNames = internalSchemaManager.getFileFieldNames(fileSchema, fieldNames);
- fieldTypes = castMap.getFileFieldTypes();
- }
- HoodieParquetReader reader = new HoodieParquetSplitReader(
- ParquetSplitReaderUtil.genPartColumnarRowReader(
- utcTimestamp,
- caseSensitive,
- conf,
- fieldNames,
- fieldTypes,
- partitionSpec,
- selectedFields,
- batchSize,
- path,
- splitStart,
- splitLength));
- if (castProjection.isPresent()) {
- return new HoodieParquetEvolvedSplitReader(reader, castProjection.get());
- } else {
- return reader;
+ Option<RowDataProjection> castProjection = castMap.toRowDataProjection(selectedFields);
+ ClosableIterator<RowData> itr = new ParquetSplitRecordIterator(
+ ParquetSplitReaderUtil.genPartColumnarRowReader(
+ utcTimestamp,
+ caseSensitive,
+ conf,
+ internalSchemaManager.getFileFieldNames(fileSchema, fieldNames), // the reconciled field names
+ castMap.getFileFieldTypes(), // the reconciled field types
+ partitionSpec,
+ selectedFields,
+ batchSize,
+ path,
+ splitStart,
+ splitLength));
+ if (castProjection.isPresent()) {
+ return new SchemaEvolvedRecordIterator(itr, castProjection.get());
+ } else {
+ return itr;
+ }
}
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetEvolvedSplitReader.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/SchemaEvolvedRecordIterator.java
similarity index 63%
rename from hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetEvolvedSplitReader.java
rename to hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/SchemaEvolvedRecordIterator.java
index 037a3776359..739512c7b55 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieParquetEvolvedSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/SchemaEvolvedRecordIterator.java
@@ -18,36 +18,35 @@
package org.apache.hudi.table.format;
+import org.apache.hudi.common.util.ClosableIterator;
import org.apache.hudi.util.RowDataProjection;
import org.apache.flink.table.data.RowData;
-import java.io.IOException;
-
/**
- * Decorates origin hoodie parquet reader with cast projection.
+ * Decorates origin record iterator with cast projection.
*/
-public final class HoodieParquetEvolvedSplitReader implements HoodieParquetReader {
- private final HoodieParquetReader originReader;
+public final class SchemaEvolvedRecordIterator implements ClosableIterator<RowData> {
+ private final ClosableIterator<RowData> nested;
private final RowDataProjection castProjection;
- public HoodieParquetEvolvedSplitReader(HoodieParquetReader originReader, RowDataProjection castProjection) {
- this.originReader = originReader;
+ public SchemaEvolvedRecordIterator(ClosableIterator<RowData> nested, RowDataProjection castProjection) {
+ this.nested = nested;
this.castProjection = castProjection;
}
@Override
- public boolean reachedEnd() throws IOException {
- return originReader.reachedEnd();
+ public boolean hasNext() {
+ return nested.hasNext();
}
@Override
- public RowData nextRecord() {
- return castProjection.project(originReader.nextRecord());
+ public RowData next() {
+ return castProjection.project(nested.next());
}
@Override
- public void close() throws IOException {
- originReader.close();
+ public void close() {
+ nested.close();
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java
index 453d0fee232..820424549f0 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java
@@ -20,8 +20,9 @@ package org.apache.hudi.table.format.cow;
import java.util.Comparator;
import org.apache.hudi.common.fs.FSUtils;
-import org.apache.hudi.table.format.HoodieParquetReader;
+import org.apache.hudi.common.util.ClosableIterator;
import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.table.format.RecordIterators;
import org.apache.hudi.util.DataTypeUtils;
import org.apache.flink.api.common.io.FileInputFormat;
@@ -75,7 +76,7 @@ public class CopyOnWriteInputFormat extends FileInputFormat<RowData> {
private final SerializableConfiguration conf;
private final long limit;
- private transient HoodieParquetReader reader;
+ private transient ClosableIterator<RowData> itr;
private transient long currentReadCount;
/**
@@ -128,7 +129,7 @@ public class CopyOnWriteInputFormat extends FileInputFormat<RowData> {
}
});
- this.reader = HoodieParquetReader.getReader(
+ this.itr = RecordIterators.getParquetRecordIterator(
internalSchemaManager,
utcTimestamp,
true,
@@ -276,26 +277,26 @@ public class CopyOnWriteInputFormat extends FileInputFormat<RowData> {
}
@Override
- public boolean reachedEnd() throws IOException {
+ public boolean reachedEnd() {
if (currentReadCount >= limit) {
return true;
} else {
- return reader.reachedEnd();
+ return !itr.hasNext();
}
}
@Override
public RowData nextRecord(RowData reuse) {
currentReadCount++;
- return reader.nextRecord();
+ return itr.next();
}
@Override
public void close() throws IOException {
- if (reader != null) {
- this.reader.close();
+ if (itr != null) {
+ this.itr.close();
}
- this.reader = null;
+ this.itr = null;
}
/**
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java
index 3103e27e857..d58a4cd0bfd 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java
@@ -29,12 +29,13 @@ import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.keygen.KeyGenUtils;
import org.apache.hudi.table.format.FilePathUtils;
import org.apache.hudi.table.format.FormatUtils;
-import org.apache.hudi.table.format.HoodieParquetReader;
import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.table.format.RecordIterators;
import org.apache.hudi.util.AvroToRowDataConverters;
import org.apache.hudi.util.DataTypeUtils;
import org.apache.hudi.util.RowDataProjection;
@@ -94,7 +95,7 @@ public class MergeOnReadInputFormat
/**
* Uniform iterator view for the underneath records.
*/
- private transient RecordIterator iterator;
+ private transient ClosableIterator<RowData> iterator;
// for project push down
/**
@@ -180,10 +181,10 @@ public class MergeOnReadInputFormat
this.iterator = new BaseFileOnlyFilteringIterator(
split.getInstantRange(),
this.tableState.getRequiredRowType(),
- getReader(split.getBasePath().get(), getRequiredPosWithCommitTime(this.requiredPos)));
+ getBaseFileIterator(split.getBasePath().get(), getRequiredPosWithCommitTime(this.requiredPos)));
} else {
// base file only
- this.iterator = new BaseFileOnlyIterator(getRequiredSchemaReader(split.getBasePath().get()));
+ this.iterator = getBaseFileIterator(split.getBasePath().get());
}
} else if (!split.getBasePath().isPresent()) {
// log files only
@@ -194,7 +195,7 @@ public class MergeOnReadInputFormat
}
} else if (split.getMergeType().equals(FlinkOptions.REALTIME_SKIP_MERGE)) {
this.iterator = new SkipMergeIterator(
- getRequiredSchemaReader(split.getBasePath().get()),
+ getBaseFileIterator(split.getBasePath().get()),
getLogFileIterator(split));
} else if (split.getMergeType().equals(FlinkOptions.REALTIME_PAYLOAD_COMBINE)) {
this.iterator = new MergeIterator(
@@ -209,7 +210,7 @@ public class MergeOnReadInputFormat
this.requiredPos,
this.emitDelete,
this.tableState.getOperationPos(),
- getFullSchemaReader(split.getBasePath().get()));
+ getBaseFileIteratorWithMetadata(split.getBasePath().get()));
} else {
throw new HoodieException("Unable to select an Iterator to read the Hoodie MOR File Split for "
+ "file path: " + split.getBasePath()
@@ -249,14 +250,14 @@ public class MergeOnReadInputFormat
return true;
} else {
// log file reaches end ?
- return this.iterator.reachedEnd();
+ return !this.iterator.hasNext();
}
}
@Override
public RowData nextRecord(RowData o) {
currentReadCount++;
- return this.iterator.nextRecord();
+ return this.iterator.next();
}
@Override
@@ -291,19 +292,19 @@ public class MergeOnReadInputFormat
}
}
- protected HoodieParquetReader getFullSchemaReader(String path) {
+ protected ClosableIterator<RowData> getBaseFileIteratorWithMetadata(String path) {
try {
- return getReader(path, IntStream.range(0, this.tableState.getRowType().getFieldCount()).toArray());
+ return getBaseFileIterator(path, IntStream.range(0, this.tableState.getRowType().getFieldCount()).toArray());
} catch (IOException e) {
throw new HoodieException("Get reader error for path: " + path);
}
}
- protected HoodieParquetReader getRequiredSchemaReader(String path) throws IOException {
- return getReader(path, this.requiredPos);
+ protected ClosableIterator<RowData> getBaseFileIterator(String path) throws IOException {
+ return getBaseFileIterator(path, this.requiredPos);
}
- private HoodieParquetReader getReader(String path, int[] requiredPos) throws IOException {
+ private ClosableIterator<RowData> getBaseFileIterator(String path, int[] requiredPos) throws IOException {
// generate partition specs.
LinkedHashMap<String, String> partSpec = FilePathUtils.extractPartitionKeyValues(
new org.apache.hadoop.fs.Path(path).getParent(),
@@ -325,7 +326,7 @@ public class MergeOnReadInputFormat
}
});
- return HoodieParquetReader.getReader(
+ return RecordIterators.getParquetRecordIterator(
internalSchemaManager,
this.conf.getBoolean(FlinkOptions.UTC_TIMEZONE),
true,
@@ -525,46 +526,12 @@ public class MergeOnReadInputFormat
// -------------------------------------------------------------------------
// Inner Class
// -------------------------------------------------------------------------
- private interface RecordIterator {
- boolean reachedEnd() throws IOException;
-
- RowData nextRecord();
-
- void close() throws IOException;
- }
-
- static class BaseFileOnlyIterator implements RecordIterator {
- // base file reader
- private final HoodieParquetReader reader;
-
- public BaseFileOnlyIterator(HoodieParquetReader reader) {
- this.reader = reader;
- }
-
- @Override
- public boolean reachedEnd() throws IOException {
- return this.reader.reachedEnd();
- }
-
- @Override
- public RowData nextRecord() {
- return this.reader.nextRecord();
- }
-
- @Override
- public void close() throws IOException {
- if (this.reader != null) {
- this.reader.close();
- }
- }
- }
-
/**
- * Similar with {@link BaseFileOnlyIterator} but with instant time filtering.
+ * Base record iterator with instant time filtering.
*/
- static class BaseFileOnlyFilteringIterator implements RecordIterator {
- // base file reader
- private final HoodieParquetReader reader;
+ static class BaseFileOnlyFilteringIterator implements ClosableIterator<RowData> {
+ // base file record iterator
+ private final ClosableIterator<RowData> nested;
private final InstantRange instantRange;
private final RowDataProjection projection;
@@ -573,44 +540,44 @@ public class MergeOnReadInputFormat
BaseFileOnlyFilteringIterator(
Option<InstantRange> instantRange,
RowType requiredRowType,
- HoodieParquetReader reader) {
- this.reader = reader;
+ ClosableIterator<RowData> nested) {
+ this.nested = nested;
this.instantRange = instantRange.orElse(null);
int[] positions = IntStream.range(1, 1 + requiredRowType.getFieldCount()).toArray();
projection = RowDataProjection.instance(requiredRowType, positions);
}
@Override
- public boolean reachedEnd() throws IOException {
- while (!this.reader.reachedEnd()) {
- currentRecord = this.reader.nextRecord();
+ public boolean hasNext() {
+ while (this.nested.hasNext()) {
+ currentRecord = this.nested.next();
if (instantRange != null) {
boolean isInRange = instantRange.isInRange(currentRecord.getString(HOODIE_COMMIT_TIME_COL_POS).toString());
if (isInRange) {
- return false;
+ return true;
}
} else {
- return false;
+ return true;
}
}
- return true;
+ return false;
}
@Override
- public RowData nextRecord() {
+ public RowData next() {
// can promote: no need to project with null instant range
return projection.project(currentRecord);
}
@Override
- public void close() throws IOException {
- if (this.reader != null) {
- this.reader.close();
+ public void close() {
+ if (this.nested != null) {
+ this.nested.close();
}
}
}
- static class LogFileOnlyIterator implements RecordIterator {
+ protected static class LogFileOnlyIterator implements ClosableIterator<RowData> {
// iterator for log files
private final ClosableIterator<RowData> iterator;
@@ -619,12 +586,12 @@ public class MergeOnReadInputFormat
}
@Override
- public boolean reachedEnd() {
- return !this.iterator.hasNext();
+ public boolean hasNext() {
+ return this.iterator.hasNext();
}
@Override
- public RowData nextRecord() {
+ public RowData next() {
return this.iterator.next();
}
@@ -636,9 +603,9 @@ public class MergeOnReadInputFormat
}
}
- static class SkipMergeIterator implements RecordIterator {
- // base file reader
- private final HoodieParquetReader reader;
+ static class SkipMergeIterator implements ClosableIterator<RowData> {
+ // base file record iterator
+ private final ClosableIterator<RowData> nested;
// iterator for log files
private final ClosableIterator<RowData> iterator;
@@ -649,34 +616,34 @@ public class MergeOnReadInputFormat
private RowData currentRecord;
- SkipMergeIterator(HoodieParquetReader reader, ClosableIterator<RowData> iterator) {
- this.reader = reader;
+ SkipMergeIterator(ClosableIterator<RowData> nested, ClosableIterator<RowData> iterator) {
+ this.nested = nested;
this.iterator = iterator;
}
@Override
- public boolean reachedEnd() throws IOException {
- if (!readLogs && !this.reader.reachedEnd()) {
- currentRecord = this.reader.nextRecord();
- return false;
+ public boolean hasNext() {
+ if (!readLogs && this.nested.hasNext()) {
+ currentRecord = this.nested.next();
+ return true;
}
readLogs = true;
if (this.iterator.hasNext()) {
currentRecord = this.iterator.next();
- return false;
+ return true;
}
- return true;
+ return false;
}
@Override
- public RowData nextRecord() {
+ public RowData next() {
return currentRecord;
}
@Override
- public void close() throws IOException {
- if (this.reader != null) {
- this.reader.close();
+ public void close() {
+ if (this.nested != null) {
+ this.nested.close();
}
if (this.iterator != null) {
this.iterator.close();
@@ -684,9 +651,9 @@ public class MergeOnReadInputFormat
}
}
- static class MergeIterator implements RecordIterator {
- // base file reader
- private final HoodieParquetReader reader;
+ protected static class MergeIterator implements ClosableIterator<RowData> {
+ // base file record iterator
+ private final ClosableIterator<RowData> nested;
// log keys used for merging
private final Iterator<String> logKeysIterator;
// scanner
@@ -726,12 +693,12 @@ public class MergeOnReadInputFormat
int[] requiredPos,
boolean emitDelete,
int operationPos,
- HoodieParquetReader reader) { // the reader should be with full schema
+ ClosableIterator<RowData> nested) { // the iterator should be with full schema
this(flinkConf, hadoopConf, split, tableRowType, requiredRowType, tableSchema,
querySchema,
Option.of(RowDataProjection.instance(requiredRowType, requiredPos)),
Option.of(record -> buildAvroRecordBySchema(record, requiredSchema, requiredPos, new GenericRecordBuilder(requiredSchema))),
- emitDelete, operationPos, reader);
+ emitDelete, operationPos, nested);
}
public MergeIterator(
@@ -746,9 +713,9 @@ public class MergeOnReadInputFormat
Option<Function<IndexedRecord, GenericRecord>> avroProjection,
boolean emitDelete,
int operationPos,
- HoodieParquetReader reader) { // the reader should be with full schema
+ ClosableIterator<RowData> nested) { // the iterator should be with full schema
this.tableSchema = tableSchema;
- this.reader = reader;
+ this.nested = nested;
this.scanner = FormatUtils.logScanner(split, tableSchema, querySchema, flinkConf, hadoopConf);
this.payloadProps = StreamerUtil.getPayloadConfig(flinkConf).getProps();
this.logKeysIterator = scanner.getRecords().keySet().iterator();
@@ -762,9 +729,9 @@ public class MergeOnReadInputFormat
}
@Override
- public boolean reachedEnd() throws IOException {
- while (!readLogs && !this.reader.reachedEnd()) {
- currentRecord = this.reader.nextRecord();
+ public boolean hasNext() {
+ while (!readLogs && this.nested.hasNext()) {
+ currentRecord = this.nested.next();
if (instantRange != null) {
boolean isInRange = instantRange.isInRange(currentRecord.getString(HOODIE_COMMIT_TIME_COL_POS).toString());
if (!isInRange) {
@@ -786,18 +753,18 @@ public class MergeOnReadInputFormat
continue;
}
IndexedRecord avroRecord = avroProjection.isPresent()
- ? avroProjection.get().apply(mergedAvroRecord.get())
- : mergedAvroRecord.get();
+ ? avroProjection.get().apply(mergedAvroRecord.get())
+ : mergedAvroRecord.get();
this.currentRecord = (RowData) avroToRowDataConverter.convert(avroRecord);
this.currentRecord.setRowKind(rowKind);
- return false;
+ return true;
}
}
// project the full record in base with required positions
if (projection.isPresent()) {
currentRecord = projection.get().project(currentRecord);
}
- return false;
+ return true;
}
// read the logs
readLogs = true;
@@ -808,46 +775,52 @@ public class MergeOnReadInputFormat
if (insertAvroRecord.isPresent()) {
// the record is a DELETE if insertAvroRecord not present, skipping
IndexedRecord avroRecord = avroProjection.isPresent()
- ? avroProjection.get().apply(insertAvroRecord.get())
- : insertAvroRecord.get();
+ ? avroProjection.get().apply(insertAvroRecord.get())
+ : insertAvroRecord.get();
this.currentRecord = (RowData) avroToRowDataConverter.convert(avroRecord);
FormatUtils.setRowKind(this.currentRecord, insertAvroRecord.get(), this.operationPos);
- return false;
+ return true;
}
}
}
- return true;
+ return false;
}
- private Option<IndexedRecord> getInsertValue(String curKey) throws IOException {
+ private Option<IndexedRecord> getInsertValue(String curKey) {
final HoodieAvroRecord<?> record = (HoodieAvroRecord) scanner.getRecords().get(curKey);
if (!emitDelete && HoodieOperation.isDelete(record.getOperation())) {
return Option.empty();
}
- return record.getData().getInsertValue(tableSchema);
+ try {
+ return record.getData().getInsertValue(tableSchema);
+ } catch (IOException e) {
+ throw new HoodieIOException("Get insert value from payload exception", e);
+ }
}
@Override
- public RowData nextRecord() {
+ public RowData next() {
return currentRecord;
}
@Override
- public void close() throws IOException {
- if (this.reader != null) {
- this.reader.close();
+ public void close() {
+ if (this.nested != null) {
+ this.nested.close();
}
if (this.scanner != null) {
this.scanner.close();
}
}
- private Option<IndexedRecord> mergeRowWithLog(
- RowData curRow,
- String curKey) throws IOException {
+ private Option<IndexedRecord> mergeRowWithLog(RowData curRow, String curKey) {
final HoodieAvroRecord<?> record = (HoodieAvroRecord) scanner.getRecords().get(curKey);
GenericRecord historyAvroRecord = (GenericRecord) rowDataToAvroConverter.convert(tableSchema, curRow);
- return record.getData().combineAndGetUpdateValue(historyAvroRecord, tableSchema, payloadProps);
+ try {
+ return record.getData().combineAndGetUpdateValue(historyAvroRecord, tableSchema, payloadProps);
+ } catch (IOException e) {
+ throw new HoodieIOException("Merge base and delta payloads exception", e);
+ }
}
}
[hudi] 09/17: [HUDI-5334] Fix checkpoint reading for structured streaming (#7389)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit bc6d6aa35d3da3f3d5a56891b922139cea7565c8
Author: Shiyan Xu <27...@users.noreply.github.com>
AuthorDate: Wed Dec 7 14:42:57 2022 +0800
[HUDI-5334] Fix checkpoint reading for structured streaming (#7389)
---
hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java
index 6baacf1ec88..ce94c92fd45 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java
@@ -151,7 +151,7 @@ public class CommitUtils {
* @return An optional commit metadata with latest checkpoint.
*/
public static Option<HoodieCommitMetadata> getLatestCommitMetadataWithValidCheckpointInfo(HoodieTimeline timeline, String checkpointKey) {
- return (Option<HoodieCommitMetadata>) timeline.getReverseOrderedInstants().map(instant -> {
+ return (Option<HoodieCommitMetadata>) timeline.filterCompletedInstants().getReverseOrderedInstants().map(instant -> {
try {
HoodieCommitMetadata commitMetadata = HoodieCommitMetadata
.fromBytes(timeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class);
[hudi] 04/17: [MINOR] Disable the `SparkSqlCoreFlow` tests (#7368)
Posted by co...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch release-0.12.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 93d766df0c45d4547249a34f93b345a246f8107a
Author: Jon Vexler <jb...@gmail.com>
AuthorDate: Fri Dec 2 22:16:46 2022 -0500
[MINOR] Disable the `SparkSqlCoreFlow` tests (#7368)
## Change Logs
Actually disable spark-sql core flow tests
---
hudi-spark-datasource/hudi-spark/pom.xml | 3 ++
.../apache/hudi/functional/SparkSQLCoreFlow.java | 32 ++++++++++++++++++++++
.../hudi/functional/TestSparkSqlCoreFlow.scala | 3 +-
3 files changed, 36 insertions(+), 2 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/pom.xml b/hudi-spark-datasource/hudi-spark/pom.xml
index b4cdd954672..f51e21d3dac 100644
--- a/hudi-spark-datasource/hudi-spark/pom.xml
+++ b/hudi-spark-datasource/hudi-spark/pom.xml
@@ -135,6 +135,9 @@
<plugin>
<groupId>org.scalatest</groupId>
<artifactId>scalatest-maven-plugin</artifactId>
+ <configuration>
+ <tagsToExclude>org.apache.hudi.functional.SparkSQLCoreFlow</tagsToExclude>
+ </configuration>
</plugin>
<plugin>
<groupId>org.scalastyle</groupId>
diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/SparkSQLCoreFlow.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/SparkSQLCoreFlow.java
new file mode 100644
index 00000000000..9fd75cd29b9
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/SparkSQLCoreFlow.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.functional;
+
+import org.scalatest.TagAnnotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@TagAnnotation
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD, ElementType.TYPE})
+public @interface SparkSQLCoreFlow {}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkSqlCoreFlow.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkSqlCoreFlow.scala
index 1143eb5faf4..fc87a5cd676 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkSqlCoreFlow.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkSqlCoreFlow.scala
@@ -29,13 +29,12 @@ import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.hudi.HoodieSparkSqlTestBase
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings
-import org.junit.jupiter.api.Disabled
import org.scalatest.Inspectors.forAll
import java.io.File
import scala.collection.JavaConversions._
-@Disabled("HUDI-5304 Disabling to avoid CI timing out")
+@SparkSQLCoreFlow
class TestSparkSqlCoreFlow extends HoodieSparkSqlTestBase {
val colsToCompare = "timestamp, _row_key, partition_path, rider, driver, begin_lat, begin_lon, end_lat, end_lon, fare.amount, fare.currency, _hoodie_is_deleted"