yihua commented on code in PR #9717:
URL: https://github.com/apache/hudi/pull/9717#discussion_r1394831451
##########
.github/workflows/bot.yml:
##########
@@ -284,29 +294,33 @@ jobs:
matrix:
include:
- flinkProfile: 'flink1.17'
- sparkProfile: 'spark3.4'
- sparkRuntime: 'spark3.4.0'
- - flinkProfile: 'flink1.17'
- sparkProfile: 'spark3.3'
- sparkRuntime: 'spark3.3.2'
- - flinkProfile: 'flink1.16'
- sparkProfile: 'spark3.3'
- sparkRuntime: 'spark3.3.2'
- - flinkProfile: 'flink1.15'
- sparkProfile: 'spark3.3'
- sparkRuntime: 'spark3.3.1'
- - flinkProfile: 'flink1.14'
- sparkProfile: 'spark3.2'
- sparkRuntime: 'spark3.2.3'
- - flinkProfile: 'flink1.13'
- sparkProfile: 'spark3.1'
- sparkRuntime: 'spark3.1.3'
- - flinkProfile: 'flink1.14'
- sparkProfile: 'spark3.0'
- sparkRuntime: 'spark3.0.2'
- - flinkProfile: 'flink1.13'
- sparkProfile: 'spark2.4'
- sparkRuntime: 'spark2.4.8'
+ sparkProfile: 'spark3.5'
+ sparkRuntime: 'spark3.5.0'
+# - flinkProfile: 'flink1.17'
+# sparkProfile: 'spark3.4'
+# sparkRuntime: 'spark3.4.0'
+# - flinkProfile: 'flink1.17'
+# sparkProfile: 'spark3.3'
+# sparkRuntime: 'spark3.3.2'
+# - flinkProfile: 'flink1.16'
+# sparkProfile: 'spark3.3'
+# sparkRuntime: 'spark3.3.2'
+# - flinkProfile: 'flink1.15'
+# sparkProfile: 'spark3.3'
+# sparkRuntime: 'spark3.3.1'
+# - flinkProfile: 'flink1.14'
+# sparkProfile: 'spark3.2'
+# sparkRuntime: 'spark3.2.3'
+# - flinkProfile: 'flink1.13'
+# sparkProfile: 'spark3.1'
+# sparkRuntime: 'spark3.1.3'
+# - flinkProfile: 'flink1.14'
+# sparkProfile: 'spark3.0'
+# sparkRuntime: 'spark3.0.2'
+# - flinkProfile: 'flink1.13'
+# sparkProfile: 'spark2.4'
+# sparkRuntime: 'spark2.4.8'
+
Review Comment:
I've built and updated the bundle validation image
`apachehudi/hudi-ci-bundle-validation-base:flink1180hive313spark350`. It's
ready for use now.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala:
##########
@@ -214,4 +215,26 @@ trait SparkAdapter extends Serializable {
* Tries to translate a Catalyst Expression into data source Filter
*/
def translateFilter(predicate: Expression, supportNestedPredicatePushdown:
Boolean = false): Option[Filter]
+
+ /**
+ * SPARK-44353 StructType#toAttributes was removed in Spark 3.5.0
+ * Use DataTypeUtils#toAttributes for Spark 3.5+
+ */
+ def toAttributes(struct: StructType): Seq[Attribute]
+
+ /**
+ * SPARK-43039 FileIndex#PartitionDirectory refactored in Spark 3.5.0
+ */
+ def toFileStatuses(partitionDirs: Seq[PartitionDirectory]): Seq[FileStatus]
+
+ /**
+ * SPARK-43039 FileIndex#PartitionDirectory refactored in Spark 3.5.0
+ */
+ def newPartitionDirectory(internalRow: InternalRow, statuses:
Seq[FileStatus]): PartitionDirectory
+
+ /**
+ * SPARK-44531 Encoder inference moved elsewhere in Spark 3.5.0
+ * Mainly used for unit tests
+ */
+ def getEncoder(schema: StructType): ExpressionEncoder[Row]
Review Comment:
Fixed now.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -77,7 +77,16 @@ object HoodieAnalysis extends SparkAdapterSupport {
}
} else {
rules += adaptIngestionTargetLogicalRelations
- val dataSourceV2ToV1FallbackClass =
"org.apache.spark.sql.hudi.analysis.HoodieDataSourceV2ToV1Fallback"
+ val dataSourceV2ToV1FallbackClass = if (HoodieSparkUtils.isSpark3_5)
+
"org.apache.spark.sql.hudi.analysis.HoodieSpark35DataSourceV2ToV1Fallback"
+ else if (HoodieSparkUtils.isSpark3_4)
+
"org.apache.spark.sql.hudi.analysis.HoodieSpark34DataSourceV2ToV1Fallback"
+ else if (HoodieSparkUtils.isSpark3_3)
+
"org.apache.spark.sql.hudi.analysis.HoodieSpark33DataSourceV2ToV1Fallback"
+ else {
+ // Spark 3.2.x
+
"org.apache.spark.sql.hudi.analysis.HoodieSpark32DataSourceV2ToV1Fallback"
+ }
Review Comment:
Looks like we have to follow this way.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/DataFrameUtil.scala:
##########
@@ -31,7 +33,7 @@ object DataFrameUtil {
*/
def createFromInternalRows(sparkSession: SparkSession, schema:
StructType, rdd: RDD[InternalRow]): DataFrame = {
- val logicalPlan = LogicalRDD(schema.toAttributes, rdd)(sparkSession)
+ val logicalPlan =
LogicalRDD(SparkAdapterSupport.sparkAdapter.toAttributes(schema),
rdd)(sparkSession)
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
