You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/05/14 19:22:00 UTC
(hudi) branch branch-0.x updated (91e176c0ef7 -> 937513b6510)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a change to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
from 91e176c0ef7 [HUDI-7431] Add replication and block size to StoragePathInfo to be backwards compatible (#10717)
new eccd183a3d3 [HUDI-7452] Repartition row dataset in S3/GCS based on task size (#10777)
new 1360b821ac6 [HUDI-7456] Set 'hudi' as the explicit provider for new table properties when create table by spark (#10776)
new dfd40dc7d8f [HUDI-7385] Add config for custom write support for parquet row writer (#10598)
new 29a2a6ccd4b Revert "[HUDI-6438] Config parameter 'MAKE_NEW_COLUMNS_NULLABLE' to allow for marking a newly created column as nullable." (#10782)
new a598bd555f1 [HUDI-7459] Update hudi-gcp-bundle pom (#10790)
new 7a5719e0660 [HUDI-7462] Refactor checkTopicCheckpoint in KafkaOffsetGen for reusability (#10794)
new 0127db94e12 [HUDI-7464] Fix minor bugs in kafka post-processing related code (#10772)
new 937513b6510 [MINOR] Fix violations of Sonarqube rule java:S2184 (#10444)
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../hudi/index/hbase/SparkHoodieHBaseIndex.java | 2 +-
.../io/storage/HoodieSparkFileWriterFactory.java | 19 ++++---
.../row/HoodieInternalRowFileWriterFactory.java | 4 +-
.../storage/row/HoodieRowParquetWriteSupport.java | 19 +++++--
.../hbase/TestHBasePutBatchSizeCalculator.java | 12 ++--
.../hudi/common/config/HoodieCommonConfig.java | 9 ---
.../hudi/common/config/HoodieStorageConfig.java | 11 ++++
.../hudi/common/fs/SizeAwareDataOutputStream.java | 2 +-
.../schema/utils/AvroSchemaEvolutionUtils.java | 9 +--
hudi-gcp/pom.xml | 2 +-
.../scala/org/apache/hudi/DataSourceOptions.scala | 2 -
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 2 +-
.../scala/org/apache/hudi/HoodieWriterUtils.scala | 1 -
.../hudi/command/CreateHoodieTableCommand.scala | 7 ++-
.../row/TestHoodieInternalRowParquetWriter.java | 3 +-
.../apache/hudi/functional/TestCOWDataSource.scala | 47 +---------------
.../apache/spark/sql/hudi/TestCreateTable.scala | 3 +-
.../hudi/utilities/HoodieWithTimelineServer.java | 2 +-
.../utilities/schema/KafkaOffsetPostProcessor.java | 35 ++++++++----
.../hudi/utilities/sources/JsonKafkaSource.java | 4 +-
.../helpers/CloudObjectsSelectorCommon.java | 15 ++++-
.../utilities/sources/helpers/KafkaOffsetGen.java | 23 ++++----
.../schema/TestKafkaOffsetPostProcessor.java | 65 ++++++++++++++++++++++
.../utilities/sources/TestJsonKafkaSource.java | 3 +
.../helpers/TestCloudObjectsSelectorCommon.java | 27 ++++++++-
.../state=CA => country=IND/state=TS}/data.json | 0
.../country=US/{state=CA => state=TX}/data.json | 0
packaging/hudi-gcp-bundle/pom.xml | 2 +-
pom.xml | 1 +
29 files changed, 208 insertions(+), 123 deletions(-)
create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestKafkaOffsetPostProcessor.java
copy hudi-utilities/src/test/resources/data/partitioned/{country=US/state=CA => country=IND/state=TS}/data.json (100%)
copy hudi-utilities/src/test/resources/data/partitioned/country=US/{state=CA => state=TX}/data.json (100%)
(hudi) 03/08: [HUDI-7385] Add config for custom write support for parquet row writer (#10598)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit dfd40dc7d8fe651fe54e282f1b057eba612c0da7
Author: Jon Vexler <jb...@gmail.com>
AuthorDate: Thu Feb 29 18:17:35 2024 -0500
[HUDI-7385] Add config for custom write support for parquet row writer (#10598)
Co-authored-by: Jonathan Vexler <=>
---
.../hudi/io/storage/HoodieSparkFileWriterFactory.java | 19 +++++++++++--------
.../row/HoodieInternalRowFileWriterFactory.java | 4 ++--
.../io/storage/row/HoodieRowParquetWriteSupport.java | 19 +++++++++++++++----
.../hudi/common/config/HoodieStorageConfig.java | 11 +++++++++++
.../row/TestHoodieInternalRowParquetWriter.java | 3 ++-
5 files changed, 41 insertions(+), 15 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
index 5feefa3bee2..7091c2b240f 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
@@ -34,6 +34,7 @@ import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.spark.sql.HoodieInternalRowUtils;
+import org.apache.spark.sql.types.StructType;
import java.io.IOException;
@@ -44,15 +45,13 @@ public class HoodieSparkFileWriterFactory extends HoodieFileWriterFactory {
String instantTime, Path path, Configuration conf, HoodieConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
boolean populateMetaFields = config.getBooleanOrDefault(HoodieTableConfig.POPULATE_META_FIELDS);
- Option<BloomFilter> filter = enableBloomFilter(populateMetaFields, config) ? Option.of(createBloomFilter(config)) : Option.empty();
String compressionCodecName = config.getStringOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME);
// Support PARQUET_COMPRESSION_CODEC_NAME is ""
if (compressionCodecName.isEmpty()) {
compressionCodecName = null;
}
- HoodieRowParquetWriteSupport writeSupport = new HoodieRowParquetWriteSupport(conf,
- HoodieInternalRowUtils.getCachedSchema(schema), filter,
- HoodieStorageConfig.newBuilder().fromProperties(config.getProps()).build());
+ HoodieRowParquetWriteSupport writeSupport = getHoodieRowParquetWriteSupport(conf, schema,
+ config, enableBloomFilter(populateMetaFields, config));
HoodieRowParquetConfig parquetConfig = new HoodieRowParquetConfig(writeSupport,
CompressionCodecName.fromConf(compressionCodecName),
config.getIntOrDefault(HoodieStorageConfig.PARQUET_BLOCK_SIZE),
@@ -69,10 +68,7 @@ public class HoodieSparkFileWriterFactory extends HoodieFileWriterFactory {
protected HoodieFileWriter newParquetFileWriter(
FSDataOutputStream outputStream, Configuration conf, HoodieConfig config, Schema schema) throws IOException {
boolean enableBloomFilter = false;
- Option<BloomFilter> filter = enableBloomFilter ? Option.of(createBloomFilter(config)) : Option.empty();
- HoodieRowParquetWriteSupport writeSupport = new HoodieRowParquetWriteSupport(conf,
- HoodieInternalRowUtils.getCachedSchema(schema), filter,
- HoodieStorageConfig.newBuilder().fromProperties(config.getProps()).build());
+ HoodieRowParquetWriteSupport writeSupport = getHoodieRowParquetWriteSupport(conf, schema, config, enableBloomFilter);
String compressionCodecName = config.getStringOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME);
// Support PARQUET_COMPRESSION_CODEC_NAME is ""
if (compressionCodecName.isEmpty()) {
@@ -100,4 +96,11 @@ public class HoodieSparkFileWriterFactory extends HoodieFileWriterFactory {
TaskContextSupplier taskContextSupplier) throws IOException {
throw new HoodieIOException("Not support write to Orc file");
}
+
+ private static HoodieRowParquetWriteSupport getHoodieRowParquetWriteSupport(Configuration conf, Schema schema,
+ HoodieConfig config, boolean enableBloomFilter) {
+ Option<BloomFilter> filter = enableBloomFilter ? Option.of(createBloomFilter(config)) : Option.empty();
+ StructType structType = HoodieInternalRowUtils.getCachedSchema(schema);
+ return HoodieRowParquetWriteSupport.getHoodieRowParquetWriteSupport(conf, structType, filter, config);
+ }
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java
index 8a61c7c44d9..ad362d17014 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java
@@ -66,8 +66,8 @@ public class HoodieInternalRowFileWriterFactory {
Option<BloomFilter> bloomFilterOpt
)
throws IOException {
- HoodieRowParquetWriteSupport writeSupport =
- new HoodieRowParquetWriteSupport(table.getHadoopConf(), structType, bloomFilterOpt, writeConfig.getStorageConfig());
+ HoodieRowParquetWriteSupport writeSupport = HoodieRowParquetWriteSupport
+ .getHoodieRowParquetWriteSupport(table.getHadoopConf(), structType, bloomFilterOpt, writeConfig);
return new HoodieInternalRowParquetWriter(
path,
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
index 3a1b6d000be..99102c30922 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
@@ -21,8 +21,11 @@ package org.apache.hudi.io.storage.row;
import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.avro.HoodieBloomFilterWriteSupport;
import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
+
import org.apache.parquet.hadoop.api.WriteSupport;
import org.apache.spark.sql.execution.datasources.parquet.ParquetWriteSupport;
import org.apache.spark.sql.types.StructType;
@@ -41,11 +44,11 @@ public class HoodieRowParquetWriteSupport extends ParquetWriteSupport {
private final Configuration hadoopConf;
private final Option<HoodieBloomFilterWriteSupport<UTF8String>> bloomFilterWriteSupportOpt;
- public HoodieRowParquetWriteSupport(Configuration conf, StructType structType, Option<BloomFilter> bloomFilterOpt, HoodieStorageConfig config) {
+ public HoodieRowParquetWriteSupport(Configuration conf, StructType structType, Option<BloomFilter> bloomFilterOpt, HoodieConfig config) {
Configuration hadoopConf = new Configuration(conf);
- hadoopConf.set("spark.sql.parquet.writeLegacyFormat", config.getString(HoodieStorageConfig.PARQUET_WRITE_LEGACY_FORMAT_ENABLED));
- hadoopConf.set("spark.sql.parquet.outputTimestampType", config.getString(HoodieStorageConfig.PARQUET_OUTPUT_TIMESTAMP_TYPE));
- hadoopConf.set("spark.sql.parquet.fieldId.write.enabled", config.getString(PARQUET_FIELD_ID_WRITE_ENABLED));
+ hadoopConf.set("spark.sql.parquet.writeLegacyFormat", config.getStringOrDefault(HoodieStorageConfig.PARQUET_WRITE_LEGACY_FORMAT_ENABLED));
+ hadoopConf.set("spark.sql.parquet.outputTimestampType", config.getStringOrDefault(HoodieStorageConfig.PARQUET_OUTPUT_TIMESTAMP_TYPE));
+ hadoopConf.set("spark.sql.parquet.fieldId.write.enabled", config.getStringOrDefault(PARQUET_FIELD_ID_WRITE_ENABLED));
setSchema(structType, hadoopConf);
this.hadoopConf = hadoopConf;
@@ -89,4 +92,12 @@ public class HoodieRowParquetWriteSupport extends ParquetWriteSupport {
}
}
+ public static HoodieRowParquetWriteSupport getHoodieRowParquetWriteSupport(Configuration conf, StructType structType,
+ Option<BloomFilter> bloomFilterOpt, HoodieConfig config) {
+ return (HoodieRowParquetWriteSupport) ReflectionUtils.loadClass(
+ config.getStringOrDefault(HoodieStorageConfig.HOODIE_PARQUET_SPARK_ROW_WRITE_SUPPORT_CLASS),
+ new Class<?>[] {Configuration.class, StructType.class, Option.class, HoodieConfig.class},
+ conf, structType, bloomFilterOpt, config);
+ }
+
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
index d68b8326ca8..f3ad183def4 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
@@ -226,6 +226,17 @@ public class HoodieStorageConfig extends HoodieConfig {
+ "and it is loaded at runtime. This is only required when trying to "
+ "override the existing write context.");
+ public static final ConfigProperty<String> HOODIE_PARQUET_SPARK_ROW_WRITE_SUPPORT_CLASS = ConfigProperty
+ .key("hoodie.parquet.spark.row.write.support.class")
+ .defaultValue("org.apache.hudi.io.storage.row.HoodieRowParquetWriteSupport")
+ .markAdvanced()
+ .sinceVersion("0.15.0")
+ .withDocumentation("Provided write support class should extend HoodieRowParquetWriteSupport class "
+ + "and it is loaded at runtime. This is only required when trying to "
+ + "override the existing write context when `hoodie.datasource.write.row.writer.enable=true`.");
+
+
+
/**
* @deprecated Use {@link #PARQUET_MAX_FILE_SIZE} and its methods instead
*/
diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java
index fb455926312..0e4dc22b8ce 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java
+++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java
@@ -131,6 +131,7 @@ public class TestHoodieInternalRowParquetWriter extends HoodieSparkClientTestHar
writeConfig.getBloomFilterFPP(),
writeConfig.getDynamicBloomFilterMaxNumEntries(),
writeConfig.getBloomFilterType());
- return new HoodieRowParquetWriteSupport(hadoopConf, SparkDatasetTestUtils.STRUCT_TYPE, Option.of(filter), writeConfig.getStorageConfig());
+ return HoodieRowParquetWriteSupport.getHoodieRowParquetWriteSupport(hadoopConf,
+ SparkDatasetTestUtils.STRUCT_TYPE, Option.of(filter), writeConfig);
}
}
(hudi) 02/08: [HUDI-7456] Set 'hudi' as the explicit provider for new table properties when create table by spark (#10776)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 1360b821ac6555c954862cc0e7f794825b360acf
Author: majian <47...@users.noreply.github.com>
AuthorDate: Thu Feb 29 19:22:52 2024 +0800
[HUDI-7456] Set 'hudi' as the explicit provider for new table properties when create table by spark (#10776)
---
.../apache/spark/sql/hudi/command/CreateHoodieTableCommand.scala | 7 ++++++-
.../src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala | 3 ++-
2 files changed, 8 insertions(+), 2 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableCommand.scala
index 3db9742aaf0..a857c3a5ded 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableCommand.scala
@@ -145,12 +145,17 @@ object CreateHoodieTableCommand {
val partitionColumnNames = hoodieCatalogTable.partitionSchema.map(_.name)
// Remove some properties should not be used;append pk, preCombineKey, type to the properties of table
- val newTblProperties =
+ var newTblProperties =
hoodieCatalogTable.catalogProperties.--(needFilterProps) ++ HoodieOptionConfig.extractSqlOptions(properties)
+
+ // Add provider -> hudi as a table property
+ newTblProperties = newTblProperties + ("provider" -> "hudi")
+
val newTable = table.copy(
identifier = newTableIdentifier,
storage = newStorage,
schema = hoodieCatalogTable.tableSchema,
+ provider = Some("hudi"),
partitionColumnNames = partitionColumnNames,
createVersion = SPARK_VERSION,
properties = newTblProperties
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala
index 937d11af6be..52290ae48b1 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala
@@ -535,9 +535,10 @@ class TestCreateTable extends HoodieSparkSqlTestBase {
)(table.schema.fields)
// Should not include non.hoodie.property
- assertResult(2)(table.properties.size)
+ assertResult(3)(table.properties.size)
assertResult("cow")(table.properties("type"))
assertResult("id,name")(table.properties("primaryKey"))
+ assertResult("hudi")(table.properties("provider"))
}
}
}
(hudi) 08/08: [MINOR] Fix violations of Sonarqube rule java:S2184 (#10444)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 937513b6510ec3b4919bbe044e81b3a659c8b6c8
Author: KUTEJiang <44...@users.noreply.github.com>
AuthorDate: Sat Mar 2 06:39:14 2024 +0800
[MINOR] Fix violations of Sonarqube rule java:S2184 (#10444)
Co-authored-by: Y Ethan Guo <et...@gmail.com>
---
.../org/apache/hudi/index/hbase/SparkHoodieHBaseIndex.java | 2 +-
.../hudi/index/hbase/TestHBasePutBatchSizeCalculator.java | 12 ++++++------
.../org/apache/hudi/common/fs/SizeAwareDataOutputStream.java | 2 +-
.../org/apache/hudi/utilities/HoodieWithTimelineServer.java | 2 +-
4 files changed, 9 insertions(+), 9 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/hbase/SparkHoodieHBaseIndex.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/hbase/SparkHoodieHBaseIndex.java
index 097e3decc2f..acbdbd2413c 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/hbase/SparkHoodieHBaseIndex.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/hbase/SparkHoodieHBaseIndex.java
@@ -551,7 +551,7 @@ public class SparkHoodieHBaseIndex extends HoodieIndex<Object, Object> {
int maxReqPerSec = getMaxReqPerSec(numRSAlive, maxQpsPerRegionServer, qpsFraction);
int numTasks = numTasksDuringPut;
int maxParallelPutsTask = Math.max(1, Math.min(numTasks, maxExecutors));
- int multiPutBatchSizePerSecPerTask = Math.max(1, (int) Math.ceil(maxReqPerSec / maxParallelPutsTask));
+ int multiPutBatchSizePerSecPerTask = Math.max(1, (int) Math.ceil((double) maxReqPerSec / maxParallelPutsTask));
LOG.info("HbaseIndexThrottling: qpsFraction :" + qpsFraction);
LOG.info("HbaseIndexThrottling: numRSAlive :" + numRSAlive);
LOG.info("HbaseIndexThrottling: maxReqPerSec :" + maxReqPerSec);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestHBasePutBatchSizeCalculator.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestHBasePutBatchSizeCalculator.java
index a6068e6a8f9..b20bc979b12 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestHBasePutBatchSizeCalculator.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestHBasePutBatchSizeCalculator.java
@@ -34,28 +34,28 @@ public class TestHBasePutBatchSizeCalculator {
int putBatchSize = batchSizeCalculator.getBatchSize(10, 16667, 1200, 200, 0.1f);
// Total puts that can be sent in 1 second = (10 * 16667 * 0.1) = 16,667
// Total puts per batch will be (16,667 / parallelism) = 83.335, where 200 is the maxExecutors
- assertEquals(putBatchSize, 83);
+ assertEquals(putBatchSize, 84);
// Number of Region Servers are halved, total requests sent in a second are also halved, so batchSize is also halved
int putBatchSize2 = batchSizeCalculator.getBatchSize(5, 16667, 1200, 200, 0.1f);
- assertEquals(putBatchSize2, 41);
+ assertEquals(putBatchSize2, 42);
// If the parallelism is halved, batchSize has to double
int putBatchSize3 = batchSizeCalculator.getBatchSize(10, 16667, 1200, 100, 0.1f);
- assertEquals(putBatchSize3, 166);
+ assertEquals(putBatchSize3, 167);
// If the parallelism is halved, batchSize has to double.
// This time parallelism is driven by numTasks rather than numExecutors
int putBatchSize4 = batchSizeCalculator.getBatchSize(10, 16667, 100, 200, 0.1f);
- assertEquals(putBatchSize4, 166);
+ assertEquals(putBatchSize4, 167);
// If sleepTimeMs is halved, batchSize has to halve
int putBatchSize5 = batchSizeCalculator.getBatchSize(10, 16667, 1200, 200, 0.05f);
- assertEquals(putBatchSize5, 41);
+ assertEquals(putBatchSize5, 42);
// If maxQPSPerRegionServer is doubled, batchSize also doubles
int putBatchSize6 = batchSizeCalculator.getBatchSize(10, 33334, 1200, 200, 0.1f);
- assertEquals(putBatchSize6, 166);
+ assertEquals(putBatchSize6, 167);
}
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/SizeAwareDataOutputStream.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/SizeAwareDataOutputStream.java
index 1cc3da6fe3c..350665d2521 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/fs/SizeAwareDataOutputStream.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/SizeAwareDataOutputStream.java
@@ -55,7 +55,7 @@ public class SizeAwareDataOutputStream {
}
public void write(byte[] v, int offset, int len) throws IOException {
- size.addAndGet(len + offset);
+ size.addAndGet((long) len + offset);
outputStream.write(v, offset, len);
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieWithTimelineServer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieWithTimelineServer.java
index e2c23b15153..fdcb806b434 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieWithTimelineServer.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieWithTimelineServer.java
@@ -102,7 +102,7 @@ public class HoodieWithTimelineServer implements Serializable {
try (CloseableHttpClient client = HttpClientBuilder.create().build()) {
System.out.println("Sleeping for " + cfg.delaySecs + " secs ");
- Thread.sleep(cfg.delaySecs * 1000);
+ Thread.sleep(cfg.delaySecs * 1000L);
System.out.println("Woke up after sleeping for " + cfg.delaySecs + " secs ");
HttpGet request = new HttpGet(url);
(hudi) 05/08: [HUDI-7459] Update hudi-gcp-bundle pom (#10790)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a598bd555f16688f07638c8e52d5ce5f2b12c86e
Author: Jinpeng <zj...@163.com>
AuthorDate: Fri Mar 1 07:31:03 2024 -0800
[HUDI-7459] Update hudi-gcp-bundle pom (#10790)
Co-authored-by: jp0317 <zj...@gmail.com>
---
hudi-gcp/pom.xml | 2 +-
packaging/hudi-gcp-bundle/pom.xml | 2 +-
pom.xml | 1 +
3 files changed, 3 insertions(+), 2 deletions(-)
diff --git a/hudi-gcp/pom.xml b/hudi-gcp/pom.xml
index 5f67569b8d2..9f078a586a5 100644
--- a/hudi-gcp/pom.xml
+++ b/hudi-gcp/pom.xml
@@ -36,7 +36,7 @@ See https://github.com/GoogleCloudPlatform/cloud-opensource-java/wiki/The-Google
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>libraries-bom</artifactId>
- <version>26.15.0</version>
+ <version>${gcp-libraries-bom.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
diff --git a/packaging/hudi-gcp-bundle/pom.xml b/packaging/hudi-gcp-bundle/pom.xml
index 4cda05d0cce..8c8f7fa72df 100644
--- a/packaging/hudi-gcp-bundle/pom.xml
+++ b/packaging/hudi-gcp-bundle/pom.xml
@@ -39,7 +39,7 @@
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>libraries-bom</artifactId>
- <version>25.1.0</version>
+ <version>${gcp-libraries-bom.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
diff --git a/pom.xml b/pom.xml
index f7660bec9a7..9158d65a890 100644
--- a/pom.xml
+++ b/pom.xml
@@ -224,6 +224,7 @@
<zookeeper.version>3.5.7</zookeeper.version>
<openjdk.jol.version>0.16</openjdk.jol.version>
<google.cloud.pubsub.version>1.120.0</google.cloud.pubsub.version>
+ <gcp-libraries-bom.version>26.15.0</gcp-libraries-bom.version>
<gcs.connector.version>hadoop2-2.2.7</gcs.connector.version>
<dynamodb-local.port>8000</dynamodb-local.port>
<dynamodb-local.endpoint>http://localhost:${dynamodb-local.port}</dynamodb-local.endpoint>
(hudi) 07/08: [HUDI-7464] Fix minor bugs in kafka post-processing related code (#10772)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 0127db94e12bb57e24ae750d26ac389f100483c9
Author: Tim Brown <ti...@onehouse.ai>
AuthorDate: Fri Mar 1 15:50:46 2024 -0600
[HUDI-7464] Fix minor bugs in kafka post-processing related code (#10772)
---
.../utilities/schema/KafkaOffsetPostProcessor.java | 35 ++++++++----
.../hudi/utilities/sources/JsonKafkaSource.java | 4 +-
.../schema/TestKafkaOffsetPostProcessor.java | 65 ++++++++++++++++++++++
.../utilities/sources/TestJsonKafkaSource.java | 3 +
4 files changed, 94 insertions(+), 13 deletions(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/KafkaOffsetPostProcessor.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/KafkaOffsetPostProcessor.java
index 500bb0c7f99..294838a435f 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/KafkaOffsetPostProcessor.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/KafkaOffsetPostProcessor.java
@@ -18,18 +18,18 @@
package org.apache.hudi.utilities.schema;
-import org.apache.avro.JsonProperties;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.internal.schema.HoodieSchemaException;
import org.apache.hudi.utilities.config.HoodieStreamerConfig;
+import org.apache.avro.JsonProperties;
import org.apache.avro.Schema;
import org.apache.spark.api.java.JavaSparkContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.Arrays;
import java.util.List;
+import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.hudi.avro.AvroSchemaUtils.createNullableSchema;
@@ -51,8 +51,6 @@ public class KafkaOffsetPostProcessor extends SchemaPostProcessor {
}
}
- private static final Logger LOG = LoggerFactory.getLogger(KafkaOffsetPostProcessor.class);
-
public static final String KAFKA_SOURCE_OFFSET_COLUMN = "_hoodie_kafka_source_offset";
public static final String KAFKA_SOURCE_PARTITION_COLUMN = "_hoodie_kafka_source_partition";
public static final String KAFKA_SOURCE_TIMESTAMP_COLUMN = "_hoodie_kafka_source_timestamp";
@@ -65,16 +63,29 @@ public class KafkaOffsetPostProcessor extends SchemaPostProcessor {
@Override
public Schema processSchema(Schema schema) {
// this method adds kafka offset fields namely source offset, partition, timestamp and kafka message key to the schema of the batch.
+ List<Schema.Field> fieldList = schema.getFields();
+ Set<String> fieldNames = fieldList.stream().map(Schema.Field::name).collect(Collectors.toSet());
+ // if the source schema already contains the kafka offset fields, then return the schema as is.
+ if (fieldNames.containsAll(Arrays.asList(KAFKA_SOURCE_OFFSET_COLUMN, KAFKA_SOURCE_PARTITION_COLUMN, KAFKA_SOURCE_TIMESTAMP_COLUMN, KAFKA_SOURCE_KEY_COLUMN))) {
+ return schema;
+ }
try {
- List<Schema.Field> fieldList = schema.getFields();
List<Schema.Field> newFieldList = fieldList.stream()
.map(f -> new Schema.Field(f.name(), f.schema(), f.doc(), f.defaultVal())).collect(Collectors.toList());
- newFieldList.add(new Schema.Field(KAFKA_SOURCE_OFFSET_COLUMN, Schema.create(Schema.Type.LONG), "offset column", 0));
- newFieldList.add(new Schema.Field(KAFKA_SOURCE_PARTITION_COLUMN, Schema.create(Schema.Type.INT), "partition column", 0));
- newFieldList.add(new Schema.Field(KAFKA_SOURCE_TIMESTAMP_COLUMN, Schema.create(Schema.Type.LONG), "timestamp column", 0));
- newFieldList.add(new Schema.Field(KAFKA_SOURCE_KEY_COLUMN, createNullableSchema(Schema.Type.STRING), "kafka key column", JsonProperties.NULL_VALUE));
- Schema newSchema = Schema.createRecord(schema.getName() + "_processed", schema.getDoc(), schema.getNamespace(), false, newFieldList);
- return newSchema;
+ // handle case where source schema provider may have already set 1 or more of these fields
+ if (!fieldNames.contains(KAFKA_SOURCE_OFFSET_COLUMN)) {
+ newFieldList.add(new Schema.Field(KAFKA_SOURCE_OFFSET_COLUMN, Schema.create(Schema.Type.LONG), "offset column", 0));
+ }
+ if (!fieldNames.contains(KAFKA_SOURCE_PARTITION_COLUMN)) {
+ newFieldList.add(new Schema.Field(KAFKA_SOURCE_PARTITION_COLUMN, Schema.create(Schema.Type.INT), "partition column", 0));
+ }
+ if (!fieldNames.contains(KAFKA_SOURCE_TIMESTAMP_COLUMN)) {
+ newFieldList.add(new Schema.Field(KAFKA_SOURCE_TIMESTAMP_COLUMN, Schema.create(Schema.Type.LONG), "timestamp column", 0));
+ }
+ if (!fieldNames.contains(KAFKA_SOURCE_KEY_COLUMN)) {
+ newFieldList.add(new Schema.Field(KAFKA_SOURCE_KEY_COLUMN, createNullableSchema(Schema.Type.STRING), "kafka key column", JsonProperties.NULL_VALUE));
+ }
+ return Schema.createRecord(schema.getName() + "_processed", schema.getDoc(), schema.getNamespace(), false, newFieldList);
} catch (Exception e) {
throw new HoodieSchemaException("Kafka offset post processor failed with schema: " + schema, e);
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
index 6e95a315260..c8c3b3421c6 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
@@ -93,7 +93,9 @@ public class JsonKafkaSource extends KafkaSource<String> {
jsonNode.put(KAFKA_SOURCE_OFFSET_COLUMN, consumerRecord.offset());
jsonNode.put(KAFKA_SOURCE_PARTITION_COLUMN, consumerRecord.partition());
jsonNode.put(KAFKA_SOURCE_TIMESTAMP_COLUMN, consumerRecord.timestamp());
- jsonNode.put(KAFKA_SOURCE_KEY_COLUMN, recordKey);
+ if (recordKey != null) {
+ jsonNode.put(KAFKA_SOURCE_KEY_COLUMN, recordKey);
+ }
stringList.add(om.writeValueAsString(jsonNode));
} catch (Throwable e) {
stringList.add(recordValue);
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestKafkaOffsetPostProcessor.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestKafkaOffsetPostProcessor.java
new file mode 100644
index 00000000000..aac441609ca
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestKafkaOffsetPostProcessor.java
@@ -0,0 +1,65 @@
+/*
+ * 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.schema;
+
+import org.apache.avro.Schema;
+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.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class TestKafkaOffsetPostProcessor {
+ private static final List<String>
+ EXPECTED_FIELD_NAMES = Arrays.asList("existing_field", "_hoodie_kafka_source_offset", "_hoodie_kafka_source_partition", "_hoodie_kafka_source_timestamp", "_hoodie_kafka_source_key");
+
+ @ParameterizedTest
+ @MethodSource("cases")
+ void testProcessSchema(Schema inputSchema) {
+ KafkaOffsetPostProcessor kafkaOffsetPostProcessor = new KafkaOffsetPostProcessor(null, null);
+ Schema actual = kafkaOffsetPostProcessor.processSchema(inputSchema);
+ List<String> actualFieldNames = actual.getFields().stream().map(Schema.Field::name).collect(Collectors.toList());
+ assertEquals(EXPECTED_FIELD_NAMES, actualFieldNames);
+ }
+
+ private static Stream<Arguments> cases() {
+ String offsetField = "{\"name\": \"_hoodie_kafka_source_offset\", \"type\": \"long\", \"doc\": \"offset column\", \"default\": 0}";
+ String partitionField = "{\"name\": \"_hoodie_kafka_source_partition\", \"type\": \"int\", \"doc\": \"partition column\", \"default\": 0}";
+ String timestampField = "{\"name\": \"_hoodie_kafka_source_timestamp\", \"type\": \"long\", \"doc\": \"timestamp column\", \"default\": 0}";
+ String keyField = "{\"name\": \"_hoodie_kafka_source_key\", \"type\": [\"null\", \"string\"], \"doc\": \"kafka key column\", \"default\": null}";
+ return Stream.of(
+ Arguments.of(new Schema.Parser().parse("{\"type\": \"record\", \"name\": \"example\", \"fields\": [{\"name\": \"existing_field\", \"type\": \"string\"}]}")),
+ Arguments.of(new Schema.Parser().parse("{\"type\": \"record\", \"name\": \"example\", \"fields\": [{\"name\": \"existing_field\", \"type\": \"string\"}, "
+ + offsetField + "]}")),
+ Arguments.of(new Schema.Parser().parse("{\"type\": \"record\", \"name\": \"example\", \"fields\": [{\"name\": \"existing_field\", \"type\": \"string\"}, "
+ + offsetField + ", " + partitionField + "]}")),
+ Arguments.of(
+ new Schema.Parser().parse("{\"type\": \"record\", \"name\": \"example\", \"fields\": [{\"name\": \"existing_field\", \"type\": \"string\"}, "
+ + offsetField + ", " + partitionField + ", " + timestampField + "]}")),
+ Arguments.of(
+ new Schema.Parser().parse("{\"type\": \"record\", \"name\": \"example\", \"fields\": [{\"name\": \"existing_field\", \"type\": \"string\"}, "
+ + offsetField + ", " + partitionField + ", " + timestampField + ", " + keyField + "]}"))
+ );
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
index 166d419001d..398c509d8e0 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
@@ -352,7 +352,10 @@ public class TestJsonKafkaSource extends BaseTestKafkaSource {
jsonSource = new JsonKafkaSource(props, jsc(), spark(), schemaProvider, metrics);
kafkaSource = new SourceFormatAdapter(jsonSource);
Dataset<Row> dfWithOffsetInfoAndNullKafkaKey = kafkaSource.fetchNewDataInRowFormat(Option.empty(), Long.MAX_VALUE).getBatch().get().cache();
+ // total of 2 * numMessages are in the topic at this point, half with a key and half with a null key. All should have the source offset.
assertEquals(numMessages, dfWithOffsetInfoAndNullKafkaKey.toDF().filter("_hoodie_kafka_source_key is null").count());
+ assertEquals(numMessages, dfWithOffsetInfoAndNullKafkaKey.toDF().filter("_hoodie_kafka_source_key is not null").count());
+ assertEquals(numMessages * 2, dfWithOffsetInfoAndNullKafkaKey.toDF().filter("_hoodie_kafka_source_offset is not null").count());
dfNoOffsetInfo.unpersist();
dfWithOffsetInfo.unpersist();
(hudi) 06/08: [HUDI-7462] Refactor checkTopicCheckpoint in KafkaOffsetGen for reusability (#10794)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 7a5719e0660ecf99e3e6ae8d3f5e3bd9f2286bf3
Author: Sampan S Nayak <sa...@gmail.com>
AuthorDate: Sat Mar 2 01:38:26 2024 +0530
[HUDI-7462] Refactor checkTopicCheckpoint in KafkaOffsetGen for reusability (#10794)
---
.../utilities/sources/helpers/KafkaOffsetGen.java | 23 +++++++++++-----------
1 file changed, 11 insertions(+), 12 deletions(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
index 57f5d38dd7c..9b1f8674ca8 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
@@ -59,6 +59,7 @@ import static org.apache.hudi.common.util.ConfigUtils.checkRequiredProperties;
import static org.apache.hudi.common.util.ConfigUtils.getBooleanWithAltKeys;
import static org.apache.hudi.common.util.ConfigUtils.getLongWithAltKeys;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.CheckpointUtils.checkTopicCheckpoint;
/**
* Source to read data from Kafka, incrementally.
@@ -68,16 +69,14 @@ public class KafkaOffsetGen {
private static final Logger LOG = LoggerFactory.getLogger(KafkaOffsetGen.class);
private static final String METRIC_NAME_KAFKA_DELAY_COUNT = "kafkaDelayCount";
private static final Comparator<OffsetRange> SORT_BY_PARTITION = Comparator.comparing(OffsetRange::partition);
-
public static final String KAFKA_CHECKPOINT_TYPE_TIMESTAMP = "timestamp";
- /**
- * kafka checkpoint Pattern.
- * Format: topic_name,partition_num:offset,partition_num:offset,....
- */
- private final Pattern pattern = Pattern.compile(".*,.*:.*");
-
public static class CheckpointUtils {
+ /**
+ * kafka checkpoint Pattern.
+ * Format: topic_name,partition_num:offset,partition_num:offset,....
+ */
+ private static final Pattern PATTERN = Pattern.compile(".*,.*:.*");
/**
* Reconstruct checkpoint from timeline.
@@ -210,6 +209,11 @@ public class KafkaOffsetGen {
public static long totalNewMessages(OffsetRange[] ranges) {
return Arrays.stream(ranges).mapToLong(OffsetRange::count).sum();
}
+
+ public static boolean checkTopicCheckpoint(Option<String> lastCheckpointStr) {
+ Matcher matcher = PATTERN.matcher(lastCheckpointStr.get());
+ return matcher.matches();
+ }
}
private final Map<String, Object> kafkaParams;
@@ -425,11 +429,6 @@ public class KafkaOffsetGen {
return result.containsKey(topicName);
}
- private boolean checkTopicCheckpoint(Option<String> lastCheckpointStr) {
- Matcher matcher = pattern.matcher(lastCheckpointStr.get());
- return matcher.matches();
- }
-
public String getTopicName() {
return topicName;
}
(hudi) 01/08: [HUDI-7452] Repartition row dataset in S3/GCS based on task size (#10777)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit eccd183a3d3f5bbf02e32032b49cfbb6aa5201a3
Author: Vinish Reddy <vi...@gmail.com>
AuthorDate: Thu Feb 29 09:01:10 2024 +0530
[HUDI-7452] Repartition row dataset in S3/GCS based on task size (#10777)
---
.../helpers/CloudObjectsSelectorCommon.java | 15 ++++++++++--
.../helpers/TestCloudObjectsSelectorCommon.java | 27 ++++++++++++++++++++--
.../partitioned/country=IND/state=TS/data.json | 1 +
.../data/partitioned/country=US/state=TX/data.json | 1 +
4 files changed, 40 insertions(+), 4 deletions(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
index 750d619258e..5ed7dcae897 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
@@ -18,7 +18,6 @@
package org.apache.hudi.utilities.sources.helpers;
-import org.apache.avro.Schema;
import org.apache.hudi.AvroConversionUtils;
import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.config.TypedProperties;
@@ -33,6 +32,7 @@ import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.InputBatch;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -204,7 +204,6 @@ public class CloudObjectsSelectorCommon {
} else {
dataset = reader.load(paths.toArray(new String[cloudObjectMetadata.size()]));
}
- dataset = dataset.coalesce(numPartitions);
// add partition column from source path if configured
if (containsConfigProperty(props, PATH_BASED_PARTITION_FIELDS)) {
@@ -216,9 +215,21 @@ public class CloudObjectsSelectorCommon {
dataset = dataset.withColumn(partitionKey, split(split(input_file_name(), partitionPathPattern).getItem(1), "/").getItem(0));
}
}
+ dataset = coalesceOrRepartition(dataset, numPartitions);
return Option.of(dataset);
}
+ private static Dataset<Row> coalesceOrRepartition(Dataset dataset, int numPartitions) {
+ int existingNumPartitions = dataset.rdd().getNumPartitions();
+ LOG.info(String.format("existing number of partitions=%d, required number of partitions=%d", existingNumPartitions, numPartitions));
+ if (existingNumPartitions < numPartitions) {
+ dataset = dataset.repartition(numPartitions);
+ } else {
+ dataset = dataset.coalesce(numPartitions);
+ }
+ return dataset;
+ }
+
public static Option<Dataset<Row>> loadAsDataset(SparkSession spark, List<CloudObjectMetadata> cloudObjectMetadata, TypedProperties props, String fileFormat) {
return loadAsDataset(spark, cloudObjectMetadata, props, fileFormat, Option.empty());
}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
index b4b6507e074..b97e2fa80a0 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
@@ -21,18 +21,19 @@ package org.apache.hudi.utilities.sources.helpers;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
-
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
+
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
-
import org.apache.spark.sql.RowFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
public class TestCloudObjectsSelectorCommon extends HoodieSparkClientTestHarness {
@@ -104,4 +105,26 @@ public class TestCloudObjectsSelectorCommon extends HoodieSparkClientTestHarness
Row expected = RowFactory.create("some data", null);
Assertions.assertEquals(Collections.singletonList(expected), result.get().collectAsList());
}
+
+ @Test
+ public void loadDatasetWithSchemaAndRepartition() {
+ TypedProperties props = new TypedProperties();
+ TestCloudObjectsSelectorCommon.class.getClassLoader().getResource("schema/sample_data_schema.avsc");
+ String schemaFilePath = TestCloudObjectsSelectorCommon.class.getClassLoader().getResource("schema/sample_data_schema.avsc").getPath();
+ props.put("hoodie.deltastreamer.schemaprovider.source.schema.file", schemaFilePath);
+ props.put("hoodie.deltastreamer.schema.provider.class.name", FilebasedSchemaProvider.class.getName());
+ props.put("hoodie.deltastreamer.source.cloud.data.partition.fields.from.path", "country,state");
+ // Setting this config so that dataset repartition happens inside `loadAsDataset`
+ props.put("hoodie.streamer.source.cloud.data.partition.max.size", "1");
+ List<CloudObjectMetadata> input = Arrays.asList(
+ new CloudObjectMetadata("src/test/resources/data/partitioned/country=US/state=CA/data.json", 1000),
+ new CloudObjectMetadata("src/test/resources/data/partitioned/country=US/state=TX/data.json", 1000),
+ new CloudObjectMetadata("src/test/resources/data/partitioned/country=IND/state=TS/data.json", 1000)
+ );
+ Option<Dataset<Row>> result = CloudObjectsSelectorCommon.loadAsDataset(sparkSession, input, props, "json", Option.of(new FilebasedSchemaProvider(props, jsc)));
+ Assertions.assertTrue(result.isPresent());
+ List<Row> expected = Arrays.asList(RowFactory.create("some data", "US", "CA"), RowFactory.create("some data", "US", "TX"), RowFactory.create("some data", "IND", "TS"));
+ List<Row> actual = result.get().collectAsList();
+ Assertions.assertEquals(new HashSet<>(expected), new HashSet<>(actual));
+ }
}
diff --git a/hudi-utilities/src/test/resources/data/partitioned/country=IND/state=TS/data.json b/hudi-utilities/src/test/resources/data/partitioned/country=IND/state=TS/data.json
new file mode 100644
index 00000000000..9fb29b4dcf4
--- /dev/null
+++ b/hudi-utilities/src/test/resources/data/partitioned/country=IND/state=TS/data.json
@@ -0,0 +1 @@
+{"data": "some data"}
\ No newline at end of file
diff --git a/hudi-utilities/src/test/resources/data/partitioned/country=US/state=TX/data.json b/hudi-utilities/src/test/resources/data/partitioned/country=US/state=TX/data.json
new file mode 100644
index 00000000000..9fb29b4dcf4
--- /dev/null
+++ b/hudi-utilities/src/test/resources/data/partitioned/country=US/state=TX/data.json
@@ -0,0 +1 @@
+{"data": "some data"}
\ No newline at end of file
(hudi) 04/08: Revert "[HUDI-6438] Config parameter 'MAKE_NEW_COLUMNS_NULLABLE' to allow for marking a newly created column as nullable." (#10782)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 29a2a6ccd4bb58353b4addd7044ac7f49b00027a
Author: Geser Dugarov, PhD <ge...@gmail.com>
AuthorDate: Fri Mar 1 07:48:04 2024 +0700
Revert "[HUDI-6438] Config parameter 'MAKE_NEW_COLUMNS_NULLABLE' to allow for marking a newly created column as nullable." (#10782)
---
.../hudi/common/config/HoodieCommonConfig.java | 9 -----
.../schema/utils/AvroSchemaEvolutionUtils.java | 9 +----
.../scala/org/apache/hudi/DataSourceOptions.scala | 2 -
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 2 +-
.../scala/org/apache/hudi/HoodieWriterUtils.scala | 1 -
.../apache/hudi/functional/TestCOWDataSource.scala | 47 +---------------------
6 files changed, 4 insertions(+), 66 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
index 65fded08e52..afb22a4a27e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
@@ -64,15 +64,6 @@ public class HoodieCommonConfig extends HoodieConfig {
+ "This enables us, to always extend the table's schema during evolution and never lose the data (when, for "
+ "ex, existing column is being dropped in a new batch)");
- public static final ConfigProperty<Boolean> MAKE_NEW_COLUMNS_NULLABLE = ConfigProperty
- .key("hoodie.datasource.write.new.columns.nullable")
- .defaultValue(false)
- .markAdvanced()
- .sinceVersion("0.14.0")
- .withDocumentation("When a non-nullable column is added to datasource during a write operation, the write "
- + " operation will fail schema compatibility check. Set this option to true will make the newly added "
- + " column nullable to successfully complete the write operation.");
-
public static final ConfigProperty<String> SET_NULL_FOR_MISSING_COLUMNS = ConfigProperty
.key("hoodie.write.set.null.for.missing.columns")
.defaultValue("false")
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 809cd2837c7..7ca0cb7f81e 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
@@ -25,11 +25,9 @@ import org.apache.avro.Schema;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;
-import static org.apache.hudi.common.config.HoodieCommonConfig.MAKE_NEW_COLUMNS_NULLABLE;
import static org.apache.hudi.common.util.CollectionUtils.reduce;
import static org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter.convert;
@@ -136,10 +134,9 @@ public class AvroSchemaEvolutionUtils {
*
* @param sourceSchema source schema that needs reconciliation
* @param targetSchema target schema that source schema will be reconciled against
- * @param opts config options
* @return schema (based off {@code source} one) that has nullability constraints and datatypes reconciled
*/
- public static Schema reconcileSchemaRequirements(Schema sourceSchema, Schema targetSchema, Map<String, String> opts) {
+ public static Schema reconcileSchemaRequirements(Schema sourceSchema, Schema targetSchema) {
if (targetSchema.getType() == Schema.Type.NULL || targetSchema.getFields().isEmpty()) {
return sourceSchema;
}
@@ -153,14 +150,12 @@ public class AvroSchemaEvolutionUtils {
List<String> colNamesSourceSchema = sourceInternalSchema.getAllColsFullName();
List<String> colNamesTargetSchema = targetInternalSchema.getAllColsFullName();
- boolean makeNewColsNullable = "true".equals(opts.get(MAKE_NEW_COLUMNS_NULLABLE.key()));
List<String> nullableUpdateColsInSource = new ArrayList<>();
List<String> typeUpdateColsInSource = new ArrayList<>();
colNamesSourceSchema.forEach(field -> {
// handle columns that needs to be made nullable
- if ((makeNewColsNullable && !colNamesTargetSchema.contains(field))
- || colNamesTargetSchema.contains(field) && sourceInternalSchema.findField(field).isOptional() != targetInternalSchema.findField(field).isOptional()) {
+ if (colNamesTargetSchema.contains(field) && sourceInternalSchema.findField(field).isOptional() != targetInternalSchema.findField(field).isOptional()) {
nullableUpdateColsInSource.add(field);
}
// handle columns that needs type to be updated
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
index 85faccdc4d7..578f7aebaf2 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
@@ -541,8 +541,6 @@ object DataSourceWriteOptions {
val SET_NULL_FOR_MISSING_COLUMNS: ConfigProperty[String] = HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS
- val MAKE_NEW_COLUMNS_NULLABLE: ConfigProperty[java.lang.Boolean] = HoodieCommonConfig.MAKE_NEW_COLUMNS_NULLABLE
-
val SPARK_SQL_INSERT_INTO_OPERATION: ConfigProperty[String] = ConfigProperty
.key("hoodie.spark.sql.insert.into.operation")
.defaultValue(WriteOperationType.INSERT.value())
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
index ed073ce4b17..0b42dc75b54 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
@@ -206,7 +206,7 @@ object HoodieSchemaUtils {
* TODO support casing reconciliation
*/
private def canonicalizeSchema(sourceSchema: Schema, latestTableSchema: Schema, opts : Map[String, String]): Schema = {
- reconcileSchemaRequirements(sourceSchema, latestTableSchema, opts)
+ reconcileSchemaRequirements(sourceSchema, latestTableSchema)
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala
index 6e541973b91..0a4ef7a3d63 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala
@@ -82,7 +82,6 @@ object HoodieWriterUtils {
hoodieConfig.setDefaultValue(ASYNC_CLUSTERING_ENABLE)
hoodieConfig.setDefaultValue(ENABLE_ROW_WRITER)
hoodieConfig.setDefaultValue(RECONCILE_SCHEMA)
- hoodieConfig.setDefaultValue(MAKE_NEW_COLUMNS_NULLABLE)
hoodieConfig.setDefaultValue(DROP_PARTITION_COLUMNS)
hoodieConfig.setDefaultValue(KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED)
Map() ++ hoodieConfig.getProps.asScala ++ globalProps ++ DataSourceOptionsHelper.translateConfigurations(parameters)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
index cb0209de979..a28a228fd46 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
@@ -24,7 +24,7 @@ import org.apache.hudi.HoodieConversionUtils.toJavaOption
import org.apache.hudi.QuickstartUtils.{convertToStringList, getQuickstartWriteConfigs}
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.config.TimestampKeyGeneratorConfig.{TIMESTAMP_INPUT_DATE_FORMAT, TIMESTAMP_OUTPUT_DATE_FORMAT, TIMESTAMP_TIMEZONE_FORMAT, TIMESTAMP_TYPE_FIELD}
-import org.apache.hudi.common.config.{HoodieCommonConfig, HoodieMetadataConfig}
+import org.apache.hudi.common.config.HoodieMetadataConfig
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.model.{HoodieRecord, WriteOperationType}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineUtils}
@@ -38,7 +38,6 @@ import org.apache.hudi.exception.ExceptionUtil.getRootCause
import org.apache.hudi.exception.HoodieException
import org.apache.hudi.functional.CommonOptionUtils._
import org.apache.hudi.functional.TestCOWDataSource.convertColumnsToNullable
-import org.apache.hudi.hive.HiveSyncConfigHolder
import org.apache.hudi.keygen._
import org.apache.hudi.keygen.constant.KeyGeneratorOptions
import org.apache.hudi.metrics.{Metrics, MetricsReporterType}
@@ -1749,50 +1748,6 @@ class TestCOWDataSource extends HoodieSparkClientTestBase with ScalaAssertionSup
assertEquals(0, result.filter(result("id") === 1).count())
}
- /** Test case to verify MAKE_NEW_COLUMNS_NULLABLE config parameter. */
- @Test
- def testSchemaEvolutionWithNewColumn(): Unit = {
- val df1 = spark.sql("select '1' as event_id, '2' as ts, '3' as version, 'foo' as event_date")
- var hudiOptions = Map[String, String](
- HoodieWriteConfig.TBL_NAME.key() -> "test_hudi_merger",
- KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key() -> "event_id",
- KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key() -> "version",
- DataSourceWriteOptions.OPERATION.key() -> "insert",
- HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key() -> "ts",
- HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key() -> "org.apache.hudi.keygen.ComplexKeyGenerator",
- KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.key() -> "true",
- HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key() -> "false",
- HoodieWriteConfig.RECORD_MERGER_IMPLS.key() -> "org.apache.hudi.HoodieSparkRecordMerger"
- )
- df1.write.format("hudi").options(hudiOptions).mode(SaveMode.Append).save(basePath)
-
- // Try adding a string column. This operation is expected to throw 'schema not compatible' exception since
- // 'MAKE_NEW_COLUMNS_NULLABLE' parameter is 'false' by default.
- val df2 = spark.sql("select '2' as event_id, '2' as ts, '3' as version, 'foo' as event_date, 'bar' as add_col")
- try {
- (df2.write.format("hudi").options(hudiOptions).mode("append").save(basePath))
- fail("Option succeeded, but was expected to fail.")
- } catch {
- case ex: org.apache.hudi.exception.HoodieInsertException => {
- assertTrue(ex.getMessage.equals("Failed insert schema compatibility check"))
- }
- case ex: Exception => {
- fail(ex)
- }
- }
-
- // Try adding the string column again. This operation is expected to succeed since 'MAKE_NEW_COLUMNS_NULLABLE'
- // parameter has been set to 'true'.
- hudiOptions = hudiOptions + (HoodieCommonConfig.MAKE_NEW_COLUMNS_NULLABLE.key() -> "true")
- try {
- (df2.write.format("hudi").options(hudiOptions).mode("append").save(basePath))
- } catch {
- case ex: Exception => {
- fail(ex)
- }
- }
- }
-
def assertLastCommitIsUpsert(): Boolean = {
val metaClient = HoodieTableMetaClient.builder()
.setBasePath(basePath)