You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2024/05/29 14:53:47 UTC
(hudi) branch branch-0.x updated: [HUDI-7809] Use Spark SerializableConfiguration to avoid NPE in Kryo serde (#11356)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-0.x by this push:
new bd26797c8e6 [HUDI-7809] Use Spark SerializableConfiguration to avoid NPE in Kryo serde (#11356)
bd26797c8e6 is described below
commit bd26797c8e651694b31bdf614f55c9fc6a757f71
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Wed May 29 07:53:40 2024 -0700
[HUDI-7809] Use Spark SerializableConfiguration to avoid NPE in Kryo serde (#11356)
* [HUDI-7809] Use Spark SerializableConfiguration to avoid NPE in Kryo serde
* Revert changes in HoodieBaseRelation
---
.../test/java/org/apache/hudi/ColumnStatsIndexHelper.java | 7 +++----
.../parquet/Spark30LegacyHoodieParquetFileFormat.scala | 12 ++++++------
.../parquet/Spark31LegacyHoodieParquetFileFormat.scala | 12 ++++++------
.../parquet/Spark32LegacyHoodieParquetFileFormat.scala | 12 ++++++------
.../parquet/Spark33LegacyHoodieParquetFileFormat.scala | 12 ++++++------
.../parquet/Spark34LegacyHoodieParquetFileFormat.scala | 12 ++++++------
.../parquet/Spark35LegacyHoodieParquetFileFormat.scala | 12 ++++++------
7 files changed, 39 insertions(+), 40 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/ColumnStatsIndexHelper.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/ColumnStatsIndexHelper.java
index 357200f5f0e..269a83bf7ac 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/ColumnStatsIndexHelper.java
+++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/ColumnStatsIndexHelper.java
@@ -21,9 +21,7 @@ import org.apache.hudi.common.model.HoodieColumnRangeMetadata;
import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hudi.util.JavaScalaConverters;
@@ -51,6 +49,7 @@ import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.types.StructType$;
import org.apache.spark.sql.types.TimestampType;
+import org.apache.spark.util.SerializableConfiguration;
import javax.annotation.Nonnull;
@@ -163,7 +162,7 @@ public class ColumnStatsIndexHelper {
.map(StructField::name)
.collect(Collectors.toList());
- StorageConfiguration<?> storageConf = HadoopFSUtils.getStorageConfWithCopy(sc.hadoopConfiguration());
+ SerializableConfiguration serializableConfiguration = new SerializableConfiguration(sc.hadoopConfiguration());
int numParallelism = (baseFilesPaths.size() / 3 + 1);
String previousJobDescription = sc.getLocalProperty("spark.job.description");
@@ -178,7 +177,7 @@ public class ColumnStatsIndexHelper {
Iterable<String> iterable = () -> paths;
return StreamSupport.stream(iterable.spliterator(), false)
.flatMap(path -> {
- HoodieStorage storage = new HoodieHadoopStorage(path, storageConf);
+ HoodieStorage storage = new HoodieHadoopStorage(path, serializableConfiguration.value());
return utils.readColumnStatsFromMetadata(
storage,
new StoragePath(path),
diff --git a/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30LegacyHoodieParquetFileFormat.scala
index bf6e222b763..59fde4af02f 100644
--- a/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30LegacyHoodieParquetFileFormat.scala
@@ -23,7 +23,6 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.util.InternalSchemaCache
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
import org.apache.hudi.common.util.collection.Pair
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.internal.schema.action.InternalSchemaMerger
import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
@@ -49,6 +48,7 @@ import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedF
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
+import org.apache.spark.util.SerializableConfiguration
import java.net.URI
@@ -108,8 +108,8 @@ class Spark30LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
hadoopConf.set(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA, prunedInternalSchemaStr)
}
- val broadcastedStorageConf =
- sparkSession.sparkContext.broadcast(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+ val broadcastedHadoopConf =
+ sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
// TODO: if you move this into the closure it reverts to the default values.
// If true, enable using the custom RecordReader for parquet. This only works for
@@ -147,7 +147,7 @@ class Spark30LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
Array.empty,
null)
- val sharedConf = broadcastedStorageConf.value.unwrap
+ val sharedConf = broadcastedHadoopConf.value.value
// Fetch internal schema
val internalSchemaStr = sharedConf.get(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA)
@@ -160,7 +160,7 @@ class Spark30LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val fileSchema = if (shouldUseInternalSchema) {
val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong;
val validCommits = sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
- val storage = new HoodieHadoopStorage(tablePath, broadcastedStorageConf.value)
+ val storage = new HoodieHadoopStorage(tablePath, sharedConf)
InternalSchemaCache.getInternalSchemaByVersionId(
commitInstantTime, tablePath, storage, if (validCommits == null) "" else validCommits)
} else {
@@ -223,7 +223,7 @@ class Spark30LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
// Clone new conf
- val hadoopAttemptConf = broadcastedStorageConf.value.unwrapCopy
+ val hadoopAttemptConf = new Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = if (shouldUseInternalSchema) {
val mergedInternalSchema = new InternalSchemaMerger(fileSchema, querySchemaOption.get(), true, true).mergeSchema()
diff --git a/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31LegacyHoodieParquetFileFormat.scala
index aa1b798241c..729ba95b644 100644
--- a/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31LegacyHoodieParquetFileFormat.scala
@@ -23,7 +23,6 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.util.InternalSchemaCache
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
import org.apache.hudi.common.util.collection.Pair
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.internal.schema.action.InternalSchemaMerger
import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
@@ -49,6 +48,7 @@ import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedF
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
+import org.apache.spark.util.SerializableConfiguration
import java.net.URI
@@ -108,8 +108,8 @@ class Spark31LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
hadoopConf.set(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA, prunedInternalSchemaStr)
}
- val broadcastedStorageConf =
- sparkSession.sparkContext.broadcast(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+ val broadcastedHadoopConf =
+ sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
// TODO: if you move this into the closure it reverts to the default values.
// If true, enable using the custom RecordReader for parquet. This only works for
@@ -147,7 +147,7 @@ class Spark31LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
Array.empty,
null)
- val sharedConf = broadcastedStorageConf.value.unwrap
+ val sharedConf = broadcastedHadoopConf.value.value
// Fetch internal schema
val internalSchemaStr = sharedConf.get(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA)
@@ -160,7 +160,7 @@ class Spark31LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val fileSchema = if (shouldUseInternalSchema) {
val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong;
val validCommits = sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
- val storage = new HoodieHadoopStorage(tablePath, broadcastedStorageConf.value)
+ val storage = new HoodieHadoopStorage(tablePath, sharedConf)
InternalSchemaCache.getInternalSchemaByVersionId(
commitInstantTime, tablePath, storage, if (validCommits == null) "" else validCommits)
} else {
@@ -227,7 +227,7 @@ class Spark31LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
// Clone new conf
- val hadoopAttemptConf = new Configuration(broadcastedStorageConf.value.unwrap)
+ val hadoopAttemptConf = new Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = if (shouldUseInternalSchema) {
val mergedInternalSchema = new InternalSchemaMerger(fileSchema, querySchemaOption.get(), true, true).mergeSchema()
diff --git a/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32LegacyHoodieParquetFileFormat.scala
index 44d420c7501..68188c3fbf0 100644
--- a/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32LegacyHoodieParquetFileFormat.scala
@@ -23,7 +23,6 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.util.InternalSchemaCache
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
import org.apache.hudi.common.util.collection.Pair
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.internal.schema.action.InternalSchemaMerger
import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
@@ -49,6 +48,7 @@ import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedF
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
+import org.apache.spark.util.SerializableConfiguration
import java.net.URI
@@ -111,8 +111,8 @@ class Spark32LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
hadoopConf.set(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA, prunedInternalSchemaStr)
}
- val broadcastedStorageConf =
- sparkSession.sparkContext.broadcast(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+ val broadcastedHadoopConf =
+ sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
// TODO: if you move this into the closure it reverts to the default values.
// If true, enable using the custom RecordReader for parquet. This only works for
@@ -146,7 +146,7 @@ class Spark32LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val filePath = new Path(new URI(file.filePath))
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
- val sharedConf = broadcastedStorageConf.value.unwrap
+ val sharedConf = broadcastedHadoopConf.value.value
// Fetch internal schema
val internalSchemaStr = sharedConf.get(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA)
@@ -159,7 +159,7 @@ class Spark32LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val fileSchema = if (shouldUseInternalSchema) {
val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong;
val validCommits = sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
- val storage = new HoodieHadoopStorage(tablePath, broadcastedStorageConf.value)
+ val storage = new HoodieHadoopStorage(tablePath, sharedConf)
InternalSchemaCache.getInternalSchemaByVersionId(
commitInstantTime, tablePath, storage, if (validCommits == null) "" else validCommits)
} else {
@@ -228,7 +228,7 @@ class Spark32LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
// Clone new conf
- val hadoopAttemptConf = broadcastedStorageConf.value.unwrapCopy
+ val hadoopAttemptConf = new Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = if (shouldUseInternalSchema) {
val mergedInternalSchema = new InternalSchemaMerger(fileSchema, querySchemaOption.get(), true, true).mergeSchema()
val mergedSchema = SparkInternalSchemaConverter.constructSparkSchemaFromInternalSchema(mergedInternalSchema)
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33LegacyHoodieParquetFileFormat.scala
index d39d12b3fe2..2e779100df3 100644
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33LegacyHoodieParquetFileFormat.scala
@@ -25,7 +25,6 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.util.InternalSchemaCache
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
import org.apache.hudi.common.util.collection.Pair
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.internal.schema.action.InternalSchemaMerger
import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
@@ -51,6 +50,7 @@ import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedF
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
+import org.apache.spark.util.SerializableConfiguration
import java.net.URI
import scala.collection.convert.ImplicitConversions.`collection AsScalaIterable`
@@ -114,8 +114,8 @@ class Spark33LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
hadoopConf.set(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA, prunedInternalSchemaStr)
}
- val broadcastedStorageConf =
- sparkSession.sparkContext.broadcast(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+ val broadcastedHadoopConf =
+ sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
// TODO: if you move this into the closure it reverts to the default values.
// If true, enable using the custom RecordReader for parquet. This only works for
@@ -148,7 +148,7 @@ class Spark33LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val filePath = new Path(new URI(file.filePath))
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
- val sharedConf = broadcastedStorageConf.value.unwrap
+ val sharedConf = broadcastedHadoopConf.value.value
// Fetch internal schema
val internalSchemaStr = sharedConf.get(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA)
@@ -161,7 +161,7 @@ class Spark33LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val fileSchema = if (shouldUseInternalSchema) {
val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong;
val validCommits = sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
- val storage = new HoodieHadoopStorage(tablePath, broadcastedStorageConf.value)
+ val storage = new HoodieHadoopStorage(tablePath, sharedConf)
InternalSchemaCache.getInternalSchemaByVersionId(
commitInstantTime, tablePath, storage, if (validCommits == null) "" else validCommits)
} else {
@@ -230,7 +230,7 @@ class Spark33LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
// Clone new conf
- val hadoopAttemptConf = broadcastedStorageConf.value.unwrapCopy
+ val hadoopAttemptConf = new Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = if (shouldUseInternalSchema) {
val mergedInternalSchema = new InternalSchemaMerger(fileSchema, querySchemaOption.get(), true, true).mergeSchema()
val mergedSchema = SparkInternalSchemaConverter.constructSparkSchemaFromInternalSchema(mergedInternalSchema)
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34LegacyHoodieParquetFileFormat.scala
index 8818cb5672f..995ef165fc4 100644
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34LegacyHoodieParquetFileFormat.scala
@@ -23,7 +23,6 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.util.InternalSchemaCache
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
import org.apache.hudi.common.util.collection.Pair
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.internal.schema.action.InternalSchemaMerger
import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
@@ -49,6 +48,7 @@ import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedF
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
+import org.apache.spark.util.SerializableConfiguration
import scala.collection.convert.ImplicitConversions.`collection AsScalaIterable`
@@ -124,8 +124,8 @@ class Spark34LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
hadoopConf.set(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA, prunedInternalSchemaStr)
}
- val broadcastedStorageConf =
- sparkSession.sparkContext.broadcast(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+ val broadcastedHadoopConf =
+ sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
// TODO: if you move this into the closure it reverts to the default values.
// If true, enable using the custom RecordReader for parquet. This only works for
@@ -160,7 +160,7 @@ class Spark34LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val filePath = file.filePath.toPath
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
- val sharedConf = broadcastedStorageConf.value.unwrap
+ val sharedConf = broadcastedHadoopConf.value.value
// Fetch internal schema
val internalSchemaStr = sharedConf.get(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA)
@@ -173,7 +173,7 @@ class Spark34LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val fileSchema = if (shouldUseInternalSchema) {
val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong;
val validCommits = sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
- val storage = new HoodieHadoopStorage(tablePath, broadcastedStorageConf.value)
+ val storage = new HoodieHadoopStorage(tablePath, sharedConf)
InternalSchemaCache.getInternalSchemaByVersionId(commitInstantTime, tablePath, storage, if (validCommits == null) "" else validCommits)
} else {
null
@@ -241,7 +241,7 @@ class Spark34LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
// Clone new conf
- val hadoopAttemptConf = broadcastedStorageConf.value.unwrapCopy
+ val hadoopAttemptConf = new Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = if (shouldUseInternalSchema) {
val mergedInternalSchema = new InternalSchemaMerger(fileSchema, querySchemaOption.get(), true, true).mergeSchema()
val mergedSchema = SparkInternalSchemaConverter.constructSparkSchemaFromInternalSchema(mergedInternalSchema)
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35LegacyHoodieParquetFileFormat.scala
index 6286a19f080..e1a3dc1427d 100644
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35LegacyHoodieParquetFileFormat.scala
@@ -23,7 +23,6 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.util.InternalSchemaCache
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
import org.apache.hudi.common.util.collection.Pair
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.internal.schema.action.InternalSchemaMerger
import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
@@ -50,6 +49,7 @@ import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedF
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
+import org.apache.spark.util.SerializableConfiguration
import scala.collection.convert.ImplicitConversions.`collection AsScalaIterable`
@@ -125,8 +125,8 @@ class Spark35LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
hadoopConf.set(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA, prunedInternalSchemaStr)
}
- val broadcastedStorageConf =
- sparkSession.sparkContext.broadcast(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+ val broadcastedHadoopConf =
+ sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
// TODO: if you move this into the closure it reverts to the default values.
// If true, enable using the custom RecordReader for parquet. This only works for
@@ -161,7 +161,7 @@ class Spark35LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val filePath = file.filePath.toPath
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
- val sharedConf = broadcastedStorageConf.value.unwrap
+ val sharedConf = broadcastedHadoopConf.value.value
// Fetch internal schema
val internalSchemaStr = sharedConf.get(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA)
@@ -174,7 +174,7 @@ class Spark35LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val fileSchema = if (shouldUseInternalSchema) {
val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong;
val validCommits = sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
- val storage = new HoodieHadoopStorage(tablePath, broadcastedStorageConf.value)
+ val storage = new HoodieHadoopStorage(tablePath, sharedConf)
InternalSchemaCache.getInternalSchemaByVersionId(
commitInstantTime, tablePath, storage, if (validCommits == null) "" else validCommits)
} else {
@@ -243,7 +243,7 @@ class Spark35LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
// Clone new conf
- val hadoopAttemptConf = broadcastedStorageConf.value.unwrapCopy
+ val hadoopAttemptConf = new Configuration(broadcastedHadoopConf.value.value)
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = if (shouldUseInternalSchema) {
val mergedInternalSchema = new InternalSchemaMerger(fileSchema, querySchemaOption.get(), true, true).mergeSchema()
val mergedSchema = SparkInternalSchemaConverter.constructSparkSchemaFromInternalSchema(mergedInternalSchema)