comphead commented on code in PR #1621:
URL: https://github.com/apache/datafusion-comet/pull/1621#discussion_r2036316165
##########
spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala:
##########
@@ -236,16 +171,70 @@ class CometSparkSessionExtensions
CometScanExec(scanExec, session)
} else {
withInfo(scanExec, fallbackReasons.mkString(", "))
- scanExec
}
case _ =>
withInfo(scanExec, s"Unsupported relation ${scanExec.relation}")
- scanExec
}
}
}
+ private def transformV2Scan(scanExec: BatchScanExec): SparkPlan = {
+
+ scanExec.scan match {
+ case scan: ParquetScan =>
+ val fallbackReasons = new ListBuffer[String]()
+ if (!CometBatchScanExec.isSchemaSupported(scan.readDataSchema)) {
+ fallbackReasons += s"Schema ${scan.readDataSchema} is not supported"
+ }
+
+ if (!CometBatchScanExec.isSchemaSupported(scan.readPartitionSchema)) {
+ fallbackReasons += s"Partition schema ${scan.readPartitionSchema} is
not supported"
+ }
+
+ if (scan.pushedAggregate.nonEmpty) {
+ fallbackReasons += "Comet does not support pushed aggregate"
+ }
+
+ if (fallbackReasons.isEmpty) {
+ val cometScan =
CometParquetScan(scanExec.scan.asInstanceOf[ParquetScan])
+ logInfo("Comet extension enabled for Scan")
+ CometBatchScanExec(
+ scanExec.copy(scan = cometScan),
+ runtimeFilters = scanExec.runtimeFilters)
+ } else {
+ withInfo(scanExec, fallbackReasons.mkString(", "))
+ }
+
+ // Iceberg scan
+ case s: SupportsComet =>
+ val fallbackReasons = new ListBuffer[String]()
+
+ if (!s.isCometEnabled) {
+ fallbackReasons += "Comet extension is not enabled for " +
+ s"${scanExec.scan.getClass.getSimpleName}: not enabled on data
source side"
+ }
+
+ if (!CometBatchScanExec.isSchemaSupported(scanExec.scan.readSchema()))
{
+ fallbackReasons += "Comet extension is not enabled for " +
+ s"${scanExec.scan.getClass.getSimpleName}: Schema not supported"
+ }
+
+ if (fallbackReasons.isEmpty) {
+ // When reading from Iceberg, we automatically enable type promotion
+ SQLConf.get.setConfString(COMET_SCHEMA_EVOLUTION_ENABLED.key, "true")
+ CometBatchScanExec(
+ scanExec.clone().asInstanceOf[BatchScanExec],
+ runtimeFilters = scanExec.runtimeFilters)
+ } else {
+ withInfo(scanExec, fallbackReasons.mkString(", "))
+ }
+
+ case _ =>
+ withInfo(scanExec, "Comet Scan only supports Parquet and Iceberg")
Review Comment:
```suggestion
withInfo(scanExec, "Comet Scan only supports Parquet and Iceberg
Parquet file formats")
```
this might be misleading, the Iceberg supports Parquet, Avro, ORC
https://iceberg.apache.org/spec/#version-1-analytic-data-tables
--
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]
