This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 4f2a80d5 feat: Implement CAST from struct to string (#1066)
4f2a80d5 is described below
commit 4f2a80d5c4ff47b60e2bfba66e7336a70ecd1e87
Author: Andy Grove <[email protected]>
AuthorDate: Fri Nov 8 08:46:56 2024 -0700
feat: Implement CAST from struct to string (#1066)
* rough out solution based on to_json
* simple test passes
* cover more types
* format
* improve checks
* update
* update
* update
* add comment
* add link to issue
---
docs/source/user-guide/compatibility.md | 1 +
docs/source/user-guide/configs.md | 32 +++----
native/spark-expr/src/cast.rs | 97 ++++++++++++++++++++--
.../org/apache/comet/expressions/CometCast.scala | 26 +++++-
.../scala/org/apache/comet/CometCastSuite.scala | 35 +++++++-
5 files changed, 162 insertions(+), 29 deletions(-)
diff --git a/docs/source/user-guide/compatibility.md
b/docs/source/user-guide/compatibility.md
index 031322e9..10459d4d 100644
--- a/docs/source/user-guide/compatibility.md
+++ b/docs/source/user-guide/compatibility.md
@@ -123,6 +123,7 @@ The following cast operations are generally compatible with
Spark except for the
| decimal | long | |
| decimal | float | |
| decimal | double | |
+| decimal | string | There can be formatting differences in some case due to
Spark using scientific notation where Comet does not |
| string | boolean | |
| string | byte | |
| string | short | |
diff --git a/docs/source/user-guide/configs.md
b/docs/source/user-guide/configs.md
index 1618a034..dc97e3c5 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -26,14 +26,14 @@ Comet provides the following configuration settings.
| spark.comet.batchSize | The columnar batch size, i.e., the maximum number of
rows that a batch can contain. | 8192 |
| spark.comet.caseConversion.enabled | Java uses locale-specific rules when
converting strings to upper or lower case and Rust does not, so we disable
upper and lower by default. | false |
| spark.comet.cast.allowIncompatible | Comet is not currently fully compatible
with Spark for all cast operations. Set this config to true to allow them
anyway. See compatibility guide for more information. | false |
-| spark.comet.columnar.shuffle.async.enabled | Whether to enable asynchronous
shuffle for Arrow-based shuffle. By default, this config is false. | false |
-| spark.comet.columnar.shuffle.async.max.thread.num | Maximum number of
threads on an executor used for Comet async columnar shuffle. By default, this
config is 100. This is the upper bound of total number of shuffle threads per
executor. In other words, if the number of cores * the number of shuffle
threads per task `spark.comet.columnar.shuffle.async.thread.num` is larger than
this config. Comet will use this config as the number of shuffle threads per
executor instead. | 100 |
-| spark.comet.columnar.shuffle.async.thread.num | Number of threads used for
Comet async columnar shuffle per shuffle task. By default, this config is 3.
Note that more threads means more memory requirement to buffer shuffle data
before flushing to disk. Also, more threads may not always improve performance,
and should be set based on the number of cores available. | 3 |
-| spark.comet.columnar.shuffle.memory.factor | Fraction of Comet memory to be
allocated per executor process for Comet shuffle. Comet memory size is
specified by `spark.comet.memoryOverhead` or calculated by
`spark.comet.memory.overhead.factor` * `spark.executor.memory`. By default,
this config is 1.0. | 1.0 |
+| spark.comet.columnar.shuffle.async.enabled | Whether to enable asynchronous
shuffle for Arrow-based shuffle. | false |
+| spark.comet.columnar.shuffle.async.max.thread.num | Maximum number of
threads on an executor used for Comet async columnar shuffle. This is the upper
bound of total number of shuffle threads per executor. In other words, if the
number of cores * the number of shuffle threads per task
`spark.comet.columnar.shuffle.async.thread.num` is larger than this config.
Comet will use this config as the number of shuffle threads per executor
instead. | 100 |
+| spark.comet.columnar.shuffle.async.thread.num | Number of threads used for
Comet async columnar shuffle per shuffle task. Note that more threads means
more memory requirement to buffer shuffle data before flushing to disk. Also,
more threads may not always improve performance, and should be set based on the
number of cores available. | 3 |
+| spark.comet.columnar.shuffle.memory.factor | Fraction of Comet memory to be
allocated per executor process for Comet shuffle. Comet memory size is
specified by `spark.comet.memoryOverhead` or calculated by
`spark.comet.memory.overhead.factor` * `spark.executor.memory`. | 1.0 |
| spark.comet.convert.csv.enabled | When enabled, data from Spark (non-native)
CSV v1 and v2 scans will be converted to Arrow format. Note that to enable
native vectorized execution, both this config and 'spark.comet.exec.enabled'
need to be enabled. | false |
| spark.comet.convert.json.enabled | When enabled, data from Spark
(non-native) JSON v1 and v2 scans will be converted to Arrow format. Note that
to enable native vectorized execution, both this config and
'spark.comet.exec.enabled' need to be enabled. | false |
| spark.comet.convert.parquet.enabled | When enabled, data from Spark
(non-native) Parquet v1 and v2 scans will be converted to Arrow format. Note
that to enable native vectorized execution, both this config and
'spark.comet.exec.enabled' need to be enabled. | false |
-| spark.comet.debug.enabled | Whether to enable debug mode for Comet. By
default, this config is false. When enabled, Comet will do additional checks
for debugging purpose. For example, validating array when importing arrays from
JVM at native side. Note that these checks may be expensive in performance and
should only be enabled for debugging purpose. | false |
+| spark.comet.debug.enabled | Whether to enable debug mode for Comet. When
enabled, Comet will do additional checks for debugging purpose. For example,
validating array when importing arrays from JVM at native side. Note that these
checks may be expensive in performance and should only be enabled for debugging
purpose. | false |
| spark.comet.dppFallback.enabled | Whether to fall back to Spark for queries
that use DPP. | true |
| spark.comet.enabled | Whether to enable Comet extension for Spark. When this
is turned on, Spark will use Comet to read Parquet data source. Note that to
enable native vectorized execution, both this config and
'spark.comet.exec.enabled' need to be enabled. By default, this config is the
value of the env var `ENABLE_COMET` if set, or true otherwise. | true |
| spark.comet.exceptionOnDatetimeRebase | Whether to throw exception when
seeing dates/timestamps from the legacy hybrid (Julian + Gregorian) calendar.
Since Spark 3, dates/timestamps were written according to the Proleptic
Gregorian calendar. When this is true, Comet will throw exceptions when seeing
these dates/timestamps that were written by Spark version before 3.0. If this
is false, these dates/timestamps will be read as if they were written to the
Proleptic Gregorian calendar and w [...]
@@ -42,13 +42,13 @@ Comet provides the following configuration settings.
| spark.comet.exec.broadcastHashJoin.enabled | Whether to enable
broadcastHashJoin by default. | true |
| spark.comet.exec.coalesce.enabled | Whether to enable coalesce by default. |
true |
| spark.comet.exec.collectLimit.enabled | Whether to enable collectLimit by
default. | true |
-| spark.comet.exec.enabled | Whether to enable Comet native vectorized
execution for Spark. This controls whether Spark should convert operators into
their Comet counterparts and execute them in native space. Note: each operator
is associated with a separate config in the format of
'spark.comet.exec.<operator_name>.enabled' at the moment, and both the config
and this need to be turned on, in order for the operator to be executed in
native. By default, this config is true. | true |
+| spark.comet.exec.enabled | Whether to enable Comet native vectorized
execution for Spark. This controls whether Spark should convert operators into
their Comet counterparts and execute them in native space. Note: each operator
is associated with a separate config in the format of
'spark.comet.exec.<operator_name>.enabled' at the moment, and both the config
and this need to be turned on, in order for the operator to be executed in
native. | true |
| spark.comet.exec.expand.enabled | Whether to enable expand by default. |
true |
| spark.comet.exec.filter.enabled | Whether to enable filter by default. |
true |
| spark.comet.exec.globalLimit.enabled | Whether to enable globalLimit by
default. | true |
| spark.comet.exec.hashJoin.enabled | Whether to enable hashJoin by default. |
true |
| spark.comet.exec.localLimit.enabled | Whether to enable localLimit by
default. | true |
-| spark.comet.exec.memoryFraction | The fraction of memory from Comet memory
overhead that the native memory manager can use for execution. The purpose of
this config is to set aside memory for untracked data structures, as well as
imprecise size estimation during memory acquisition. Default value is 0.7. |
0.7 |
+| spark.comet.exec.memoryFraction | The fraction of memory from Comet memory
overhead that the native memory manager can use for execution. The purpose of
this config is to set aside memory for untracked data structures, as well as
imprecise size estimation during memory acquisition. | 0.7 |
| spark.comet.exec.project.enabled | Whether to enable project by default. |
true |
| spark.comet.exec.replaceSortMergeJoin | Experimental feature to force Spark
to replace SortMergeJoin with ShuffledHashJoin for improved performance. This
feature is not stable yet. For more information, refer to the Comet Tuning
Guide (https://datafusion.apache.org/comet/user-guide/tuning.html). | false |
| spark.comet.exec.shuffle.codec | The codec of Comet native shuffle used to
compress shuffle data. Only zstd is supported. | zstd |
@@ -63,18 +63,18 @@ Comet provides the following configuration settings.
| spark.comet.explain.native.enabled | When this setting is enabled, Comet
will provide a tree representation of the native query plan before execution
and again after execution, with metrics. | false |
| spark.comet.explain.verbose.enabled | When this setting is enabled, Comet
will provide a verbose tree representation of the extended information. | false
|
| spark.comet.explainFallback.enabled | When this setting is enabled, Comet
will provide logging explaining the reason(s) why a query stage cannot be
executed natively. Set this to false to reduce the amount of logging. | false |
-| spark.comet.memory.overhead.factor | Fraction of executor memory to be
allocated as additional non-heap memory per executor process for Comet. Default
value is 0.2. | 0.2 |
+| spark.comet.memory.overhead.factor | Fraction of executor memory to be
allocated as additional non-heap memory per executor process for Comet. | 0.2 |
| spark.comet.memory.overhead.min | Minimum amount of additional memory to be
allocated per executor process for Comet, in MiB. | 402653184b |
| spark.comet.nativeLoadRequired | Whether to require Comet native library to
load successfully when Comet is enabled. If not, Comet will silently fallback
to Spark when it fails to load the native lib. Otherwise, an error will be
thrown and the Spark job will be aborted. | false |
-| spark.comet.parquet.enable.directBuffer | Whether to use Java direct byte
buffer when reading Parquet. By default, this is false | false |
-| spark.comet.parquet.read.io.adjust.readRange.skew | In the parallel reader,
if the read ranges submitted are skewed in sizes, this option will cause the
reader to break up larger read ranges into smaller ranges to reduce the skew.
This will result in a slightly larger number of connections opened to the file
system but may give improved performance. The option is off by default. | false
|
-| spark.comet.parquet.read.io.mergeRanges | When enabled the parallel reader
will try to merge ranges of data that are separated by less than
'comet.parquet.read.io.mergeRanges.delta' bytes. Longer continuous reads are
faster on cloud storage. The default behavior is to merge consecutive ranges. |
true |
+| spark.comet.parquet.enable.directBuffer | Whether to use Java direct byte
buffer when reading Parquet. | false |
+| spark.comet.parquet.read.io.adjust.readRange.skew | In the parallel reader,
if the read ranges submitted are skewed in sizes, this option will cause the
reader to break up larger read ranges into smaller ranges to reduce the skew.
This will result in a slightly larger number of connections opened to the file
system but may give improved performance. | false |
+| spark.comet.parquet.read.io.mergeRanges | When enabled the parallel reader
will try to merge ranges of data that are separated by less than
'comet.parquet.read.io.mergeRanges.delta' bytes. Longer continuous reads are
faster on cloud storage. | true |
| spark.comet.parquet.read.io.mergeRanges.delta | The delta in bytes between
consecutive read ranges below which the parallel reader will try to merge the
ranges. The default is 8MB. | 8388608 |
-| spark.comet.parquet.read.parallel.io.enabled | Whether to enable Comet's
parallel reader for Parquet files. The parallel reader reads ranges of
consecutive data in a file in parallel. It is faster for large files and row
groups but uses more resources. The parallel reader is enabled by default. |
true |
+| spark.comet.parquet.read.parallel.io.enabled | Whether to enable Comet's
parallel reader for Parquet files. The parallel reader reads ranges of
consecutive data in a file in parallel. It is faster for large files and row
groups but uses more resources. | true |
| spark.comet.parquet.read.parallel.io.thread-pool.size | The maximum number
of parallel threads the parallel reader will use in a single executor. For
executors configured with a smaller number of cores, use a smaller number. | 16
|
| spark.comet.regexp.allowIncompatible | Comet is not currently fully
compatible with Spark for all regular expressions. Set this config to true to
allow them anyway using Rust's regular expression engine. See compatibility
guide for more information. | false |
-| spark.comet.scan.enabled | Whether to enable native scans. When this is
turned on, Spark will use Comet to read supported data sources (currently only
Parquet is supported natively). Note that to enable native vectorized
execution, both this config and 'spark.comet.exec.enabled' need to be enabled.
By default, this config is true. | true |
-| spark.comet.scan.preFetch.enabled | Whether to enable pre-fetching feature
of CometScan. By default is disabled. | false |
-| spark.comet.scan.preFetch.threadNum | The number of threads running
pre-fetching for CometScan. Effective if spark.comet.scan.preFetch.enabled is
enabled. By default it is 2. Note that more pre-fetching threads means more
memory requirement to store pre-fetched row groups. | 2 |
-| spark.comet.shuffle.preferDictionary.ratio | The ratio of total values to
distinct values in a string column to decide whether to prefer dictionary
encoding when shuffling the column. If the ratio is higher than this config,
dictionary encoding will be used on shuffling string column. This config is
effective if it is higher than 1.0. By default, this config is 10.0. Note that
this config is only used when `spark.comet.exec.shuffle.mode` is `jvm`. | 10.0 |
+| spark.comet.scan.enabled | Whether to enable native scans. When this is
turned on, Spark will use Comet to read supported data sources (currently only
Parquet is supported natively). Note that to enable native vectorized
execution, both this config and 'spark.comet.exec.enabled' need to be enabled.
| true |
+| spark.comet.scan.preFetch.enabled | Whether to enable pre-fetching feature
of CometScan. | false |
+| spark.comet.scan.preFetch.threadNum | The number of threads running
pre-fetching for CometScan. Effective if spark.comet.scan.preFetch.enabled is
enabled. Note that more pre-fetching threads means more memory requirement to
store pre-fetched row groups. | 2 |
+| spark.comet.shuffle.preferDictionary.ratio | The ratio of total values to
distinct values in a string column to decide whether to prefer dictionary
encoding when shuffling the column. If the ratio is higher than this config,
dictionary encoding will be used on shuffling string column. This config is
effective if it is higher than 1.0. Note that this config is only used when
`spark.comet.exec.shuffle.mode` is `jvm`. | 10.0 |
| spark.comet.sparkToColumnar.supportedOperatorList | A comma-separated list
of operators that will be converted to Arrow columnar format when
'spark.comet.sparkToColumnar.enabled' is true | Range,InMemoryTableScan |
diff --git a/native/spark-expr/src/cast.rs b/native/spark-expr/src/cast.rs
index 0224aabf..0a2f7fef 100644
--- a/native/spark-expr/src/cast.rs
+++ b/native/spark-expr/src/cast.rs
@@ -32,8 +32,14 @@ use arrow::{
record_batch::RecordBatch,
util::display::FormatOptions,
};
-use arrow_array::DictionaryArray;
+use arrow_array::builder::StringBuilder;
+use arrow_array::{DictionaryArray, StringArray, StructArray};
use arrow_schema::{DataType, Schema};
+use datafusion_common::{
+ cast::as_generic_string_array, internal_err, Result as DataFusionResult,
ScalarValue,
+};
+use datafusion_expr::ColumnarValue;
+use datafusion_physical_expr::PhysicalExpr;
use std::str::FromStr;
use std::{
any::Any,
@@ -43,12 +49,6 @@ use std::{
sync::Arc,
};
-use datafusion_common::{
- cast::as_generic_string_array, internal_err, Result as DataFusionResult,
ScalarValue,
-};
-use datafusion_expr::ColumnarValue;
-use datafusion_physical_expr::PhysicalExpr;
-
use chrono::{NaiveDate, NaiveDateTime, TimeZone, Timelike};
use datafusion::physical_expr_common::physical_expr::down_cast_any_ref;
use num::{
@@ -711,6 +711,9 @@ fn cast_array(
{
spark_cast_nonintegral_numeric_to_integral(&array, eval_mode,
from_type, to_type)
}
+ (DataType::Struct(_), DataType::Utf8) => {
+ Ok(casts_struct_to_string(array.as_struct(), &timezone)?)
+ }
_ if is_datafusion_spark_compatible(from_type, to_type,
allow_incompat) => {
// use DataFusion cast only when we know that it is compatible
with Spark
Ok(cast_with_options(&array, to_type, &CAST_OPTIONS)?)
@@ -785,6 +788,7 @@ fn is_datafusion_spark_compatible(
| DataType::Float64
| DataType::Decimal128(_, _)
| DataType::Decimal256(_, _)
+ | DataType::Utf8 // note that there can be formatting
differences
),
DataType::Utf8 if allow_incompat => matches!(
to_type,
@@ -807,6 +811,52 @@ fn is_datafusion_spark_compatible(
}
}
+fn casts_struct_to_string(array: &StructArray, timezone: &str) ->
DataFusionResult<ArrayRef> {
+ // cast each field to a string
+ let string_arrays: Vec<ArrayRef> = array
+ .columns()
+ .iter()
+ .map(|arr| {
+ spark_cast(
+ ColumnarValue::Array(Arc::clone(arr)),
+ &DataType::Utf8,
+ EvalMode::Legacy,
+ timezone,
+ true,
+ )
+ .and_then(|cv| cv.into_array(arr.len()))
+ })
+ .collect::<DataFusionResult<Vec<_>>>()?;
+ let string_arrays: Vec<&StringArray> =
+ string_arrays.iter().map(|arr| arr.as_string()).collect();
+ // build the struct string containing entries in the format
`"field_name":field_value`
+ let mut builder = StringBuilder::with_capacity(array.len(), array.len() *
16);
+ let mut str = String::with_capacity(array.len() * 16);
+ for row_index in 0..array.len() {
+ if array.is_null(row_index) {
+ builder.append_null();
+ } else {
+ str.clear();
+ let mut any_fields_written = false;
+ str.push('{');
+ for field in &string_arrays {
+ if any_fields_written {
+ str.push_str(", ");
+ }
+ if field.is_null(row_index) {
+ str.push_str("null");
+ } else {
+ str.push_str(field.value(row_index));
+ }
+ any_fields_written = true;
+ }
+ str.push('}');
+ builder.append_value(&str);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
fn cast_string_to_int<OffsetSize: OffsetSizeTrait>(
to_type: &DataType,
array: &ArrayRef,
@@ -1879,7 +1929,7 @@ fn trim_end(s: &str) -> &str {
mod tests {
use arrow::datatypes::TimestampMicrosecondType;
use arrow_array::StringArray;
- use arrow_schema::TimeUnit;
+ use arrow_schema::{Field, TimeUnit};
use std::str::FromStr;
use super::*;
@@ -2255,4 +2305,35 @@ mod tests {
);
assert!(result.is_err())
}
+
+ #[test]
+ fn test_cast_struct_to_utf8() {
+ let a: ArrayRef = Arc::new(Int32Array::from(vec![
+ Some(1),
+ Some(2),
+ None,
+ Some(4),
+ Some(5),
+ ]));
+ let b: ArrayRef = Arc::new(StringArray::from(vec!["a", "b", "c", "d",
"e"]));
+ let c: ArrayRef = Arc::new(StructArray::from(vec![
+ (Arc::new(Field::new("a", DataType::Int32, true)), a),
+ (Arc::new(Field::new("b", DataType::Utf8, true)), b),
+ ]));
+ let string_array = cast_array(
+ c,
+ &DataType::Utf8,
+ EvalMode::Legacy,
+ "UTC".to_owned(),
+ false,
+ )
+ .unwrap();
+ let string_array = string_array.as_string::<i32>();
+ assert_eq!(5, string_array.len());
+ assert_eq!(r#"{1, a}"#, string_array.value(0));
+ assert_eq!(r#"{2, b}"#, string_array.value(1));
+ assert_eq!(r#"{null, c}"#, string_array.value(2));
+ assert_eq!(r#"{4, d}"#, string_array.value(3));
+ assert_eq!(r#"{5, e}"#, string_array.value(4));
+ }
}
diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index 9725c5f2..eb9800b8 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -19,7 +19,7 @@
package org.apache.comet.expressions
-import org.apache.spark.sql.types.{DataType, DataTypes, DecimalType}
+import org.apache.spark.sql.types.{DataType, DataTypes, DecimalType,
StructType}
sealed trait SupportLevel
@@ -76,7 +76,7 @@ object CometCast {
case (DataTypes.StringType, _) =>
canCastFromString(toType, timeZoneId, evalMode)
case (_, DataTypes.StringType) =>
- canCastToString(fromType)
+ canCastToString(fromType, timeZoneId, evalMode)
case (DataTypes.TimestampType, _) =>
canCastFromTimestamp(toType)
case (_: DecimalType, _) =>
@@ -137,7 +137,10 @@ object CometCast {
}
}
- private def canCastToString(fromType: DataType): SupportLevel = {
+ private def canCastToString(
+ fromType: DataType,
+ timeZoneId: Option[String],
+ evalMode: CometEvalMode.Value): SupportLevel = {
fromType match {
case DataTypes.BooleanType => Compatible()
case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType |
@@ -151,9 +154,26 @@ object CometCast {
"There can be differences in precision. " +
"For example, the input \"1.4E-45\" will produce 1.0E-45 " +
"instead of 1.4E-45"))
+ case _: DecimalType =>
+ // https://github.com/apache/datafusion-comet/issues/1068
+ Compatible(
+ Some(
+ "There can be formatting differences in some case due to Spark
using " +
+ "scientific notation where Comet does not"))
case DataTypes.BinaryType =>
// https://github.com/apache/datafusion-comet/issues/377
Incompatible(Some("Only works for binary data representing valid UTF-8
strings"))
+ case StructType(fields) =>
+ for (field <- fields) {
+ isSupported(field.dataType, DataTypes.StringType, timeZoneId,
evalMode) match {
+ case s: Incompatible =>
+ return s
+ case Unsupported =>
+ return Unsupported
+ case _ =>
+ }
+ }
+ Compatible()
case _ => Unsupported
}
}
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
index 4bcc3e00..817545c5 100644
--- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
@@ -24,6 +24,7 @@ import java.io.File
import scala.util.Random
import scala.util.matching.Regex
+import org.apache.hadoop.fs.Path
import org.apache.spark.sql.{CometTestBase, DataFrame, SaveMode}
import org.apache.spark.sql.catalyst.expressions.Cast
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
@@ -470,8 +471,7 @@ class CometCastSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
castTest(generateDecimalsPrecision38Scale18(), DataTypes.LongType)
}
- ignore("cast DecimalType(10,2) to StringType") {
- // input: 0E-18, expected: 0E-18, actual: 0.000000000000000000
+ test("cast DecimalType(10,2) to StringType") {
castTest(generateDecimalsPrecision10Scale2(), DataTypes.StringType)
}
@@ -850,6 +850,37 @@ class CometCastSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
castTest(generateTimestamps(), DataTypes.DateType)
}
+ // Complex Types
+
+ test("cast StructType to StringType") {
+ Seq(true, false).foreach { dictionaryEnabled =>
+ withTempDir { dir =>
+ val path = new Path(dir.toURI.toString, "test.parquet")
+ makeParquetFileAllTypes(path, dictionaryEnabled = dictionaryEnabled,
10000)
+ withParquetTable(path.toString, "tbl") {
+ // primitives
+ checkSparkAnswerAndOperator(
+ "SELECT CAST(struct(_1, _2, _3, _4, _5, _6, _7, _8) as string)
FROM tbl")
+ // TODO: enable tests for unsigned ints (_9, _10, _11, _12) once
+ // https://github.com/apache/datafusion-comet/issues/1067 is
resolved
+ // checkSparkAnswerAndOperator(
+ // "SELECT CAST(struct(_9, _10, _11, _12) as string) FROM tbl")
+ // decimals
+ // TODO add _16 when
https://github.com/apache/datafusion-comet/issues/1068 is resolved
+ checkSparkAnswerAndOperator("SELECT CAST(struct(_15, _17) as string)
FROM tbl")
+ // dates & timestamps
+ checkSparkAnswerAndOperator("SELECT CAST(struct(_18, _19, _20) as
string) FROM tbl")
+ // named struct
+ checkSparkAnswerAndOperator(
+ "SELECT CAST(named_struct('a', _1, 'b', _2) as string) FROM tbl")
+ // nested struct
+ checkSparkAnswerAndOperator(
+ "SELECT CAST(named_struct('a', named_struct('b', _1, 'c', _2)) as
string) FROM tbl")
+ }
+ }
+ }
+ }
+
private def generateFloats(): DataFrame = {
withNulls(gen.generateFloats(dataSize)).toDF("a")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
