Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove merged PR #1747: URL: https://github.com/apache/datafusion-comet/pull/1747 -- 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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on PR #1747: URL: https://github.com/apache/datafusion-comet/pull/1747#issuecomment-2971468514 Thanks for the review @parthchandra. I added the logging. -- 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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2145516448
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
Review Comment:
Yes, that is a good idea. I will add that today.
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
parthchandra commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2145482367
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
Review Comment:
Should we also log a message to indicate that auto scan was enabled (and
the scan implementation it selected)?
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
Review Comment:
I see. Thank you for the explanation :)
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2145468079
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
Review Comment:
The current implementation of `hasKnownIssues` only returns `true` for maps
~of~ containing complex types (whether at the top level of within other complex
types), because at the time of writing, there was a known issue with maps of
structs. I'm not sure if this is still the case.
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2145468079
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
Review Comment:
The current implementation of `hasKnownIssues` only returns `true` for maps
of complex types (whether at the top level of within other complex types),
because at the time of writing, there was a known issue with maps of structs.
I'm not sure if this is still the case.
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
parthchandra commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2145401316
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
Review Comment:
Right.
For `Array(elementType = Struct(PrimitiveType)`, `hasKnownIssues` would
return `false`
For `Map(..., valueType=Struct(PrimitiveType)`, `hasKnownIssues` would
return `true`
Is that what is intended? Not sure why that would be the case, tbh.
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2145246950
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
Review Comment:
I made these checks quite conservative for now. Hopefully we can relax the
rules at some point.
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2145229513
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
Review Comment:
`hasKnownIssues` would return `true` for a `Map` containing `Struct` due to
the `isComplexType` checks.
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
parthchandra commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2143805849
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -105,8 +105,49 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
Review Comment:
This seems to indicate that if we have an array type with the element type
being a struct with primitive fields then there are no known issues. But if we
have a map with a value type of the same struct with primitive fields then it
is not supported?
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2143610758
##
.github/workflows/spark_sql_test_native_auto.yml:
##
@@ -0,0 +1,71 @@
+# 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.
+
+name: Spark SQL Tests (native_auto)
+
+concurrency:
+ group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{
github.workflow }}
+ cancel-in-progress: true
+
+on:
+ # manual trigger
+ #
https://docs.github.com/en/actions/managing-workflow-runs/manually-running-a-workflow
+ workflow_dispatch:
+
+env:
+ RUST_VERSION: stable
+
+jobs:
+ spark-sql-catalyst-native-auto:
+strategy:
+ matrix:
+os: [ubuntu-24.04]
+java-version: [11]
+spark-version: [{short: '3.4', full: '3.4.3'}, {short: '3.5', full:
'3.5.5'}]
+module:
+ - {name: "catalyst", args1: "catalyst/test", args2: ""}
+ - {name: "sql/core-1", args1: "", args2: sql/testOnly * -- -l
org.apache.spark.tags.ExtendedSQLTest -l org.apache.spark.tags.SlowSQLTest}
+ - {name: "sql/core-2", args1: "", args2: "sql/testOnly * -- -n
org.apache.spark.tags.ExtendedSQLTest"}
+ - {name: "sql/core-3", args1: "", args2: "sql/testOnly * -- -n
org.apache.spark.tags.SlowSQLTest"}
+ - {name: "sql/hive-1", args1: "", args2: "hive/testOnly * -- -l
org.apache.spark.tags.ExtendedHiveTest -l org.apache.spark.tags.SlowHiveTest"}
+ - {name: "sql/hive-2", args1: "", args2: "hive/testOnly * -- -n
org.apache.spark.tags.ExtendedHiveTest"}
+ - {name: "sql/hive-3", args1: "", args2: "hive/testOnly * -- -n
org.apache.spark.tags.SlowHiveTest"}
+ fail-fast: false
+name: spark-sql-native-auto-${{ matrix.module.name }}/${{ matrix.os
}}/spark-${{ matrix.spark-version.full }}/java-${{ matrix.java-version }}
+runs-on: ${{ matrix.os }}
+container:
+ image: amd64/rust
+steps:
+ - uses: actions/checkout@v4
+ - name: Setup Rust & Java toolchain
+uses: ./.github/actions/setup-builder
+with:
+ rust-version: ${{env.RUST_VERSION}}
+ jdk-version: ${{ matrix.java-version }}
+ - name: Setup Spark
+uses: ./.github/actions/setup-spark-builder
+with:
+ spark-version: ${{ matrix.spark-version.full }}
+ spark-short-version: ${{ matrix.spark-version.short }}
+ - name: Run Spark tests
+run: |
+ cd apache-spark
+ rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet
cache requires cleanups
+ ENABLE_COMET=true ENABLE_COMET_SHUFFLE=true
COMET_PARQUET_SCAN_IMPL=auto build/sbt ${{ matrix.module.args1 }} "${{
matrix.module.args2 }}"
Review Comment:
`auto` mode specified here
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2124216384
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -93,21 +93,63 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
+}
+ }
+
+ val knownIssues =
+scanExec.requiredSchema.exists(field =>
hasKnownIssues(field.dataType)) ||
+ r.partitionSchema.exists(field => hasKnownIssues(field.dataType))
+
+ if (COMET_EXEC_ENABLED
+ .get() && schemaSupported && partitionSchemaSupported &&
+!scanExec.bucketedScan && !knownIssues) {
+scanImpl = SCAN_NATIVE_ICEBERG_COMPAT
Review Comment:
thanks, I will update this
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on code in PR #1747:
URL: https://github.com/apache/datafusion-comet/pull/1747#discussion_r2124215025
##
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##
@@ -93,21 +93,63 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] {
return withInfos(scanExec, fallbackReasons.toSet)
}
-val scanImpl = COMET_NATIVE_SCAN_IMPL.get()
-if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION &&
!COMET_EXEC_ENABLED.get()) {
+var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
+
+// if scan is auto then pick best available scan
+if (scanImpl == SCAN_AUTO) {
+ // TODO these checks are not yet exhaustive. For example,
native_datafusion does
+ // not support reading from object stores such as S3 yet
+
+ val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+ val schemaSupported =
+typeChecker.isSchemaSupported(scanExec.requiredSchema,
fallbackReasons)
+ val partitionSchemaSupported =
+typeChecker.isSchemaSupported(r.partitionSchema, fallbackReasons)
+
+ // additional checks for known issues
+ def isComplexType(dt: DataType): Boolean = dt match {
+case _: StructType | _: ArrayType | _: MapType => true
+case _ => false
+ }
+
+ def hasKnownIssues(dataType: DataType): Boolean = {
+dataType match {
+ case s: StructType => s.exists(field =>
hasKnownIssues(field.dataType))
+ case a: ArrayType => hasKnownIssues(a.elementType)
+ case m: MapType => isComplexType(m.keyType) ||
isComplexType(m.valueType)
+ case _ => false
+}
+ }
+
+ val knownIssues =
+scanExec.requiredSchema.exists(field =>
hasKnownIssues(field.dataType)) ||
+ r.partitionSchema.exists(field => hasKnownIssues(field.dataType))
+
+ if (COMET_EXEC_ENABLED
+ .get() && schemaSupported && partitionSchemaSupported &&
+!scanExec.bucketedScan && !knownIssues) {
+scanImpl = SCAN_NATIVE_ICEBERG_COMPAT
+ }
+}
+
+if (scanImpl == SCAN_AUTO) {
+ scanImpl = SCAN_NATIVE_COMET
Review Comment:
I figured that we have more Spark SQL tests passing with
`native_iceberg_compat` so we should start with that.
--
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]
Re: [PR] feat: Add experimental auto mode for `COMET_PARQUET_SCAN_IMPL` [datafusion-comet]
andygrove commented on PR #1747: URL: https://github.com/apache/datafusion-comet/pull/1747#issuecomment-2935920067 @parthchandra @mbutrovich Could I get a review? I changed the scope to adding the "auto" option without changing the default. There is a manual workflow where we can run the Spark SQL tests using the new auto mode to see which tests fail (if any). -- 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]
