[comet-parquet-exec] Move type conversion logic for ParquetExec out of Cast expression. (#1229)

* Copy cast.rs logic to a new parquet_support.rs. Remove Parquet dependencies on cast.rs.

* Move parquet_support and schema_adapter to parquet folder.

* Add fields to SparkParquetOptions.

Author: mbutrovich

native/Cargo.lock

@@ -79,6 +79,7 @@ datafusion-comet-spark-expr = { workspace = true }
 datafusion-comet-proto = { workspace = true }
 object_store = { workspace = true }
 url = { workspace = true }
+chrono = { workspace = true }
 
 [dev-dependencies]
 pprof = { version = "0.13.0", features = ["flamegraph"] }

native/core/Cargo.toml

@@ -65,13 +65,13 @@ use datafusion::{
     },
     prelude::SessionContext,
 };
-use datafusion_comet_spark_expr::{
-    create_comet_physical_fun, create_negate_expr, SparkSchemaAdapterFactory,
-};
+use datafusion_comet_spark_expr::{create_comet_physical_fun, create_negate_expr};
 use datafusion_functions_nested::concat::ArrayAppend;
 use datafusion_physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr};
 
 use crate::execution::spark_plan::SparkPlan;
+use crate::parquet::parquet_support::SparkParquetOptions;
+use crate::parquet::schema_adapter::SparkSchemaAdapterFactory;
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::physical_plan::parquet::ParquetExecBuilder;
 use datafusion::datasource::physical_plan::FileScanConfig;
@@ -1156,13 +1156,14 @@ impl PhysicalPlanner {
                 table_parquet_options.global.pushdown_filters = true;
                 table_parquet_options.global.reorder_filters = true;
 
-                let mut spark_cast_options = SparkCastOptions::new(EvalMode::Legacy, "UTC", false);
-                spark_cast_options.allow_cast_unsigned_ints = true;
+                let mut spark_parquet_options =
+                    SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+                spark_parquet_options.allow_cast_unsigned_ints = true;
 
                 let mut builder = ParquetExecBuilder::new(file_scan_config)
                     .with_table_parquet_options(table_parquet_options)
                     .with_schema_adapter_factory(Arc::new(SparkSchemaAdapterFactory::new(
-                        spark_cast_options,
+                        spark_parquet_options,
                     )));
 
                 if let Some(filter) = cnf_data_filters {

native/core/src/execution/planner.rs

@@ -21,7 +21,9 @@ pub use mutable_vector::*;
 
 #[macro_use]
 pub mod util;
+pub mod parquet_support;
 pub mod read;
+pub mod schema_adapter;
 
 use std::task::Poll;
 use std::{boxed::Box, ptr::NonNull, sync::Arc};
@@ -40,27 +42,28 @@ use jni::{
     },
 };
 
+use self::util::jni::TypePromotionInfo;
 use crate::execution::operators::ExecutionError;
 use crate::execution::utils::SparkArrowConvert;
 use crate::parquet::data_type::AsBytes;
+use crate::parquet::schema_adapter::SparkSchemaAdapterFactory;
 use arrow::buffer::{Buffer, MutableBuffer};
 use arrow_array::{Array, RecordBatch};
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::physical_plan::parquet::ParquetExecBuilder;
 use datafusion::datasource::physical_plan::FileScanConfig;
 use datafusion::physical_plan::ExecutionPlan;
-use datafusion_comet_spark_expr::{EvalMode, SparkCastOptions, SparkSchemaAdapterFactory};
+use datafusion_comet_spark_expr::EvalMode;
 use datafusion_common::config::TableParquetOptions;
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
 use futures::{poll, StreamExt};
 use jni::objects::{JBooleanArray, JByteArray, JLongArray, JPrimitiveArray, JString, ReleaseMode};
 use jni::sys::jstring;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReader;
+use parquet_support::SparkParquetOptions;
 use read::ColumnReader;
 use util::jni::{convert_column_descriptor, convert_encoding, deserialize_schema, get_file_path};
 
-use self::util::jni::TypePromotionInfo;
-
 /// Parquet read context maintained across multiple JNI calls.
 struct Context {
     pub column_reader: ColumnReader,
@@ -679,13 +682,13 @@ pub unsafe extern "system" fn Java_org_apache_comet_parquet_Native_initRecordBat
         table_parquet_options.global.pushdown_filters = true;
         table_parquet_options.global.reorder_filters = true;
 
-        let mut spark_cast_options = SparkCastOptions::new(EvalMode::Legacy, "UTC", false);
-        spark_cast_options.allow_cast_unsigned_ints = true;
+        let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+        spark_parquet_options.allow_cast_unsigned_ints = true;
 
         let builder2 = ParquetExecBuilder::new(file_scan_config)
             .with_table_parquet_options(table_parquet_options)
             .with_schema_adapter_factory(Arc::new(SparkSchemaAdapterFactory::new(
-                spark_cast_options,
+                spark_parquet_options,
             )));
 
         //TODO: (ARROW NATIVE) - predicate pushdown??