infvg commented on code in PR #12962:
URL: https://github.com/apache/gluten/pull/12962#discussion_r3926795754
##########
gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala:
##########
@@ -162,65 +157,72 @@ object GlutenIcebergSourceUtil {
metadataColumns
}
- def getFileFormat(sparkScan: Scan): ReadFileFormat = sparkScan match {
- case scan: SparkBatchQueryScan =>
- val tasks = scan.tasks().asScala
- asFileScanTask(tasks.toList).foreach {
- task =>
- task.file().format() match {
- case FileFormat.PARQUET => return ReadFileFormat.ParquetReadFormat
- case FileFormat.ORC => return ReadFileFormat.OrcReadFormat
- case _ =>
- }
- }
- throw new GlutenNotSupportException("Iceberg Only support parquet and
orc file format.")
- case _ =>
- throw new GlutenNotSupportException("Only support iceberg
SparkBatchQueryScan.")
+ def getFileFormat(sparkScan: Scan): ReadFileFormat = {
+ asFileScanTask(getScanTasks(sparkScan)).foreach {
+ task =>
+ task.file().format() match {
+ case FileFormat.PARQUET => return ReadFileFormat.ParquetReadFormat
+ case FileFormat.ORC => return ReadFileFormat.OrcReadFormat
+ case _ =>
+ }
+ }
+ throw new GlutenNotSupportException("Iceberg Only support parquet and orc
file format.")
}
- def getReadPartitionSchema(sparkScan: Scan): StructType = sparkScan match {
- case scan: SparkBatchQueryScan =>
- val tasks = scan.tasks().asScala
- asFileScanTask(tasks.toList).foreach {
- task =>
- val spec = task.spec()
- if (spec.isPartitioned) {
- val readFields = scan.readSchema().fields.map(_.name).toSet
- // Iceberg will generate some non-table fields as partition
fields, such as x_bucket,
- // which will not appear in readFields, they also cannot be
filtered.
- val tableFields =
spec.schema().columns().asScala.map(_.name()).toSet
- val voidTransformFields = scan
- .table()
- .spec()
+ def getReadPartitionSchema(sparkScan: Scan): StructType = {
+ asFileScanTask(getScanTasks(sparkScan)).foreach {
+ task =>
+ val spec = task.spec()
+ if (spec.isPartitioned) {
+ val readFields = sparkScan.readSchema().fields.map(_.name).toSet
+ // Iceberg will generate some non-table fields as partition fields,
such as x_bucket,
+ // which will not appear in readFields, they also cannot be filtered.
+ val tableFields = spec.schema().columns().asScala.map(_.name()).toSet
+ val voidTransformFields = getTable(sparkScan)
+ .spec()
Review Comment:
`task.spec()` only gives the partition spec used to write the file, we still
need `table.spec()` for the fields dropped from the current spec using void
transforms
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
