datafusion_proto/physical_plan/

from_proto.rs

// 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.

//! Serde code to convert from protocol buffers to Rust data structures.

use std::sync::Arc;

use arrow::array::RecordBatch;
use arrow::compute::SortOptions;
use arrow::datatypes::{Field, Schema};
use arrow::ipc::reader::StreamReader;
use chrono::{TimeZone, Utc};
use datafusion_common::{DataFusionError, Result, internal_datafusion_err, not_impl_err};
use datafusion_datasource::file::FileSource;
use datafusion_datasource::file_groups::FileGroup;
use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
use datafusion_datasource::file_sink_config::FileSinkConfig;
use datafusion_datasource::{FileRange, ListingTableUrl, PartitionedFile, TableSchema};
use datafusion_datasource_csv::file_format::CsvSink;
use datafusion_datasource_json::file_format::JsonSink;
#[cfg(feature = "parquet")]
use datafusion_datasource_parquet::file_format::ParquetSink;
use datafusion_execution::object_store::ObjectStoreUrl;
use datafusion_execution::{FunctionRegistry, TaskContext};
use datafusion_expr::WindowFunctionDefinition;
use datafusion_expr::dml::InsertOp;
use datafusion_expr::execution_props::SubqueryIndex;
use datafusion_physical_expr::projection::{ProjectionExpr, ProjectionExprs};
use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr;
use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr, ScalarFunctionExpr};
use datafusion_physical_plan::expressions::{
    BinaryExpr, CaseExpr, CastExpr, Column, IsNotNullExpr, IsNullExpr, LikeExpr, Literal,
    NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn, in_list,
};
use datafusion_physical_plan::joins::{HashExpr, SeededRandomState};
use datafusion_physical_plan::windows::{create_window_expr, schema_add_window_field};
use datafusion_physical_plan::{Partitioning, PhysicalExpr, WindowExpr};
use datafusion_proto_common::common::proto_error;
use object_store::ObjectMeta;
use object_store::path::Path;

use super::{
    DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
    PhysicalProtoConverterExtension,
};
use crate::logical_plan::{self};
use crate::protobuf::physical_expr_node::ExprType;
use crate::{convert_required, protobuf};
use datafusion_physical_expr::expressions::{
    DynamicFilterInner, DynamicFilterPhysicalExpr,
};

impl From<&protobuf::PhysicalColumn> for Column {
    fn from(c: &protobuf::PhysicalColumn) -> Column {
        Column::new(&c.name, c.index as usize)
    }
}

/// Parses a physical sort expression from a protobuf.
///
/// # Arguments
///
/// * `proto` - Input proto with physical sort expression node
/// * `input_schema` - The Arrow schema for the input, used for determining expression data types
///   when performing type coercion.
/// * `ctx` - Decode context carrying the task context, extension codec, and
///   any scoped state needed during recursive deserialization.
/// * `proto_converter` - Converter hooks used for recursive physical plan and
///   expression deserialization.
pub fn parse_physical_sort_expr(
    proto: &protobuf::PhysicalSortExprNode,
    ctx: &PhysicalPlanDecodeContext<'_>,
    input_schema: &Schema,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<PhysicalSortExpr> {
    if let Some(expr) = &proto.expr {
        let expr =
            proto_converter.proto_to_physical_expr(expr.as_ref(), input_schema, ctx)?;
        let options = SortOptions {
            descending: !proto.asc,
            nulls_first: proto.nulls_first,
        };
        Ok(PhysicalSortExpr { expr, options })
    } else {
        Err(proto_error("Unexpected empty physical expression"))
    }
}

/// Parses a physical sort expressions from a protobuf.
///
/// # Arguments
///
/// * `proto` - Input proto with vector of physical sort expression node
/// * `input_schema` - The Arrow schema for the input, used for determining expression data types
///   when performing type coercion.
/// * `ctx` - Decode context carrying the task context, extension codec, and
///   any scoped state needed during recursive deserialization.
/// * `proto_converter` - Converter hooks used for recursive physical plan and
///   expression deserialization.
pub fn parse_physical_sort_exprs(
    proto: &[protobuf::PhysicalSortExprNode],
    ctx: &PhysicalPlanDecodeContext<'_>,
    input_schema: &Schema,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Vec<PhysicalSortExpr>> {
    proto
        .iter()
        .map(|sort_expr| {
            parse_physical_sort_expr(sort_expr, ctx, input_schema, proto_converter)
        })
        .collect()
}

/// Parses a physical window expr from a protobuf.
///
/// # Arguments
///
/// * `proto` - Input proto with physical window expression node.
/// * `name` - Name of the window expression.
/// * `input_schema` - The Arrow schema for the input, used for determining
///   expression data types when performing type coercion.
/// * `ctx` - Decode context carrying the task context, extension codec, and
///   any scoped state needed during recursive deserialization.
/// * `proto_converter` - Converter hooks used for recursive physical plan and
///   expression deserialization.
pub fn parse_physical_window_expr(
    proto: &protobuf::PhysicalWindowExprNode,
    ctx: &PhysicalPlanDecodeContext<'_>,
    input_schema: &Schema,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn WindowExpr>> {
    let window_node_expr =
        parse_physical_exprs(&proto.args, ctx, input_schema, proto_converter)?;
    let partition_by =
        parse_physical_exprs(&proto.partition_by, ctx, input_schema, proto_converter)?;

    let order_by =
        parse_physical_sort_exprs(&proto.order_by, ctx, input_schema, proto_converter)?;

    let window_frame = proto
        .window_frame
        .as_ref()
        .map(|wf| wf.clone().try_into())
        .transpose()
        .map_err(|e| internal_datafusion_err!("{e}"))?
        .ok_or_else(|| {
            internal_datafusion_err!("Missing required field 'window_frame' in protobuf")
        })?;

    let fun = if let Some(window_func) = proto.window_function.as_ref() {
        match window_func {
            protobuf::physical_window_expr_node::WindowFunction::UserDefinedAggrFunction(udaf_name) => {
                WindowFunctionDefinition::AggregateUDF(match &proto.fun_definition {
                    Some(buf) => ctx.codec().try_decode_udaf(udaf_name, buf)?,
                    None => ctx
                        .task_ctx()
                        .udaf(udaf_name)
                        .or_else(|_| ctx.codec().try_decode_udaf(udaf_name, &[]))?,
                })
            }
            protobuf::physical_window_expr_node::WindowFunction::UserDefinedWindowFunction(udwf_name) => {
                WindowFunctionDefinition::WindowUDF(match &proto.fun_definition {
                    Some(buf) => ctx.codec().try_decode_udwf(udwf_name, buf)?,
                    None => ctx
                        .task_ctx()
                        .udwf(udwf_name)
                        .or_else(|_| ctx.codec().try_decode_udwf(udwf_name, &[]))?
                })
            }
        }
    } else {
        return Err(proto_error("Missing required field in protobuf"));
    };

    let name = proto.name.clone();
    // TODO: Remove extended_schema if functions are all UDAF
    let extended_schema =
        schema_add_window_field(&window_node_expr, input_schema, &fun, &name)?;
    create_window_expr(
        &fun,
        name,
        &window_node_expr,
        &partition_by,
        &order_by,
        Arc::new(window_frame),
        extended_schema,
        proto.ignore_nulls,
        proto.distinct,
        None,
    )
}

pub fn parse_physical_exprs<'a, I>(
    protos: I,
    ctx: &PhysicalPlanDecodeContext<'_>,
    input_schema: &Schema,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Vec<Arc<dyn PhysicalExpr>>>
where
    I: IntoIterator<Item = &'a protobuf::PhysicalExprNode>,
{
    protos
        .into_iter()
        .map(|p| proto_converter.proto_to_physical_expr(p, input_schema, ctx))
        .collect::<Result<Vec<_>>>()
}

/// Parses a physical expression from a protobuf.
///
/// # Arguments
///
/// * `proto` - Input proto with physical expression node
/// * `ctx` - Task context used to resolve registered functions.
/// * `input_schema` - The Arrow schema for the input, used for determining
///   expression data types when performing type coercion.
/// * `codec` - Physical extension codec used to construct the root decode
///   context for deserialization.
pub fn parse_physical_expr(
    proto: &protobuf::PhysicalExprNode,
    ctx: &TaskContext,
    input_schema: &Schema,
    codec: &dyn PhysicalExtensionCodec,
) -> Result<Arc<dyn PhysicalExpr>> {
    let decode_ctx = PhysicalPlanDecodeContext::new(ctx, codec);
    parse_physical_expr_with_converter(
        proto,
        input_schema,
        &decode_ctx,
        &DefaultPhysicalProtoConverter {},
    )
}

/// Parses a physical expression from a protobuf.
///
/// # Arguments
///
/// * `proto` - Input proto with physical expression node
/// * `input_schema` - The Arrow schema for the input, used for determining
///   expression data types when performing type coercion.
/// * `ctx` - Decode context carrying the task context, extension codec, and
///   any scoped state needed during recursive deserialization.
/// * `proto_converter` - Converter hooks used for recursive physical plan and
///   expression deserialization.
pub fn parse_physical_expr_with_converter(
    proto: &protobuf::PhysicalExprNode,
    input_schema: &Schema,
    ctx: &PhysicalPlanDecodeContext<'_>,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn PhysicalExpr>> {
    let expr_type = proto
        .expr_type
        .as_ref()
        .ok_or_else(|| proto_error("Unexpected empty physical expression"))?;

    let pexpr: Arc<dyn PhysicalExpr> = match expr_type {
        ExprType::Column(c) => {
            let pcol: Column = c.into();
            Arc::new(pcol)
        }
        ExprType::UnknownColumn(c) => Arc::new(UnKnownColumn::new(&c.name)),
        ExprType::Literal(scalar) => Arc::new(Literal::new(scalar.try_into()?)),
        ExprType::BinaryExpr(binary_expr) => {
            let op = logical_plan::from_proto::from_proto_binary_op(&binary_expr.op)?;
            if !binary_expr.operands.is_empty() {
                // New linearized format: reduce the flat operands list back into
                // a nested binary expression tree.
                let operands: Vec<Arc<dyn PhysicalExpr>> = binary_expr
                    .operands
                    .iter()
                    .map(|e| proto_converter.proto_to_physical_expr(e, input_schema, ctx))
                    .collect::<Result<Vec<_>>>()?;

                if operands.len() < 2 {
                    return Err(proto_error(
                        "A binary expression must always have at least 2 operands",
                    ));
                }

                operands
                    .into_iter()
                    .reduce(|left, right| Arc::new(BinaryExpr::new(left, op, right)))
                    .expect(
                        "Binary expression could not be reduced to a single expression.",
                    )
            } else {
                // Legacy format with l/r fields
                Arc::new(BinaryExpr::new(
                    parse_required_physical_expr(
                        binary_expr.l.as_deref(),
                        ctx,
                        "left",
                        input_schema,
                        proto_converter,
                    )?,
                    op,
                    parse_required_physical_expr(
                        binary_expr.r.as_deref(),
                        ctx,
                        "right",
                        input_schema,
                        proto_converter,
                    )?,
                ))
            }
        }
        ExprType::AggregateExpr(_) => {
            return not_impl_err!(
                "Cannot convert aggregate expr node to physical expression"
            );
        }
        ExprType::WindowExpr(_) => {
            return not_impl_err!(
                "Cannot convert window expr node to physical expression"
            );
        }
        ExprType::Sort(_) => {
            return not_impl_err!("Cannot convert sort expr node to physical expression");
        }
        ExprType::IsNullExpr(e) => {
            Arc::new(IsNullExpr::new(parse_required_physical_expr(
                e.expr.as_deref(),
                ctx,
                "expr",
                input_schema,
                proto_converter,
            )?))
        }
        ExprType::IsNotNullExpr(e) => {
            Arc::new(IsNotNullExpr::new(parse_required_physical_expr(
                e.expr.as_deref(),
                ctx,
                "expr",
                input_schema,
                proto_converter,
            )?))
        }
        ExprType::NotExpr(e) => Arc::new(NotExpr::new(parse_required_physical_expr(
            e.expr.as_deref(),
            ctx,
            "expr",
            input_schema,
            proto_converter,
        )?)),
        ExprType::Negative(e) => {
            Arc::new(NegativeExpr::new(parse_required_physical_expr(
                e.expr.as_deref(),
                ctx,
                "expr",
                input_schema,
                proto_converter,
            )?))
        }
        ExprType::InList(e) => in_list(
            parse_required_physical_expr(
                e.expr.as_deref(),
                ctx,
                "expr",
                input_schema,
                proto_converter,
            )?,
            parse_physical_exprs(&e.list, ctx, input_schema, proto_converter)?,
            &e.negated,
            input_schema,
        )?,
        ExprType::Case(e) => Arc::new(CaseExpr::try_new(
            e.expr
                .as_ref()
                .map(|e| {
                    proto_converter.proto_to_physical_expr(e.as_ref(), input_schema, ctx)
                })
                .transpose()?,
            e.when_then_expr
                .iter()
                .map(|e| {
                    Ok((
                        parse_required_physical_expr(
                            e.when_expr.as_ref(),
                            ctx,
                            "when_expr",
                            input_schema,
                            proto_converter,
                        )?,
                        parse_required_physical_expr(
                            e.then_expr.as_ref(),
                            ctx,
                            "then_expr",
                            input_schema,
                            proto_converter,
                        )?,
                    ))
                })
                .collect::<Result<Vec<_>>>()?,
            e.else_expr
                .as_ref()
                .map(|e| {
                    proto_converter.proto_to_physical_expr(e.as_ref(), input_schema, ctx)
                })
                .transpose()?,
        )?),
        ExprType::Cast(e) => Arc::new(CastExpr::new(
            parse_required_physical_expr(
                e.expr.as_deref(),
                ctx,
                "expr",
                input_schema,
                proto_converter,
            )?,
            convert_required!(e.arrow_type)?,
            None,
        )),
        ExprType::TryCast(e) => Arc::new(TryCastExpr::new(
            parse_required_physical_expr(
                e.expr.as_deref(),
                ctx,
                "expr",
                input_schema,
                proto_converter,
            )?,
            convert_required!(e.arrow_type)?,
        )),
        ExprType::ScalarUdf(e) => {
            let udf = match &e.fun_definition {
                Some(buf) => ctx.codec().try_decode_udf(&e.name, buf)?,
                None => ctx
                    .task_ctx()
                    .udf(e.name.as_str())
                    .or_else(|_| ctx.codec().try_decode_udf(&e.name, &[]))?,
            };
            let scalar_fun_def = Arc::clone(&udf);

            let args = parse_physical_exprs(&e.args, ctx, input_schema, proto_converter)?;

            let config_options = Arc::clone(ctx.task_ctx().session_config().options());

            Arc::new(
                ScalarFunctionExpr::new(
                    e.name.as_str(),
                    scalar_fun_def,
                    args,
                    Field::new(
                        &e.return_field_name,
                        convert_required!(e.return_type)?,
                        true,
                    )
                    .into(),
                    config_options,
                )
                .with_nullable(e.nullable),
            )
        }
        ExprType::LikeExpr(like_expr) => Arc::new(LikeExpr::new(
            like_expr.negated,
            like_expr.case_insensitive,
            parse_required_physical_expr(
                like_expr.expr.as_deref(),
                ctx,
                "expr",
                input_schema,
                proto_converter,
            )?,
            parse_required_physical_expr(
                like_expr.pattern.as_deref(),
                ctx,
                "pattern",
                input_schema,
                proto_converter,
            )?,
        )),
        ExprType::HashExpr(hash_expr) => {
            let on_columns = parse_physical_exprs(
                &hash_expr.on_columns,
                ctx,
                input_schema,
                proto_converter,
            )?;
            Arc::new(HashExpr::new(
                on_columns,
                SeededRandomState::with_seed(hash_expr.seed0),
                hash_expr.description.clone(),
            ))
        }
        ExprType::ScalarSubquery(sq) => {
            let data_type: arrow::datatypes::DataType = sq
                .data_type
                .as_ref()
                .ok_or_else(|| {
                    proto_error("Missing data_type in PhysicalScalarSubqueryExprNode")
                })?
                .try_into()?;
            let results = ctx.scalar_subquery_results().ok_or_else(|| {
                proto_error(
                    "ScalarSubqueryExpr can only be deserialized as part \
                         of a surrounding ScalarSubqueryExec",
                )
            })?;
            Arc::new(ScalarSubqueryExpr::new(
                data_type,
                sq.nullable,
                SubqueryIndex::new(sq.index as usize),
                results.clone(),
            ))
        }
        ExprType::DynamicFilter(dynamic_filter) => {
            let children = parse_physical_exprs(
                &dynamic_filter.children,
                ctx,
                input_schema,
                proto_converter,
            )?;

            let remapped_children = if !dynamic_filter.remapped_children.is_empty() {
                Some(parse_physical_exprs(
                    &dynamic_filter.remapped_children,
                    ctx,
                    input_schema,
                    proto_converter,
                )?)
            } else {
                None
            };

            let inner_expr = parse_required_physical_expr(
                dynamic_filter.inner_expr.as_deref(),
                ctx,
                "inner_expr",
                input_schema,
                proto_converter,
            )?;

            let expression_id = proto.expr_id.ok_or_else(|| {
                proto_error(
                    "DynamicFilterPhysicalExpr requires PhysicalExprNode.expr_id \
                     to be set by the serializer",
                )
            })?;

            let base_filter: Arc<dyn PhysicalExpr> =
                Arc::new(DynamicFilterPhysicalExpr::from_parts(
                    children,
                    remapped_children,
                    DynamicFilterInner {
                        expression_id,
                        generation: dynamic_filter.generation,
                        expr: inner_expr,
                        is_complete: dynamic_filter.is_complete,
                    },
                ));
            base_filter
        }
        ExprType::Extension(extension) => {
            let inputs: Vec<Arc<dyn PhysicalExpr>> = extension
                .inputs
                .iter()
                .map(|e| proto_converter.proto_to_physical_expr(e, input_schema, ctx))
                .collect::<Result<_>>()?;
            ctx.codec()
                .try_decode_expr(extension.expr.as_slice(), &inputs)? as _
        }
    };

    Ok(pexpr)
}

fn parse_required_physical_expr(
    expr: Option<&protobuf::PhysicalExprNode>,
    ctx: &PhysicalPlanDecodeContext<'_>,
    field: &str,
    input_schema: &Schema,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn PhysicalExpr>> {
    expr.map(|e| proto_converter.proto_to_physical_expr(e, input_schema, ctx))
        .transpose()?
        .ok_or_else(|| internal_datafusion_err!("Missing required field {field:?}"))
}

pub fn parse_protobuf_hash_partitioning(
    partitioning: Option<&protobuf::PhysicalHashRepartition>,
    ctx: &PhysicalPlanDecodeContext<'_>,
    input_schema: &Schema,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Option<Partitioning>> {
    match partitioning {
        Some(hash_part) => {
            let expr = parse_physical_exprs(
                &hash_part.hash_expr,
                ctx,
                input_schema,
                proto_converter,
            )?;

            Ok(Some(Partitioning::Hash(
                expr,
                hash_part.partition_count.try_into().unwrap(),
            )))
        }
        None => Ok(None),
    }
}

pub fn parse_protobuf_partitioning(
    partitioning: Option<&protobuf::Partitioning>,
    ctx: &PhysicalPlanDecodeContext<'_>,
    input_schema: &Schema,
    proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Option<Partitioning>> {
    match partitioning {
        Some(protobuf::Partitioning { partition_method }) => match partition_method {
            Some(protobuf::partitioning::PartitionMethod::RoundRobin(
                partition_count,
            )) => Ok(Some(Partitioning::RoundRobinBatch(
                *partition_count as usize,
            ))),
            Some(protobuf::partitioning::PartitionMethod::Hash(hash_repartition)) => {
                parse_protobuf_hash_partitioning(
                    Some(hash_repartition),
                    ctx,
                    input_schema,
                    proto_converter,
                )
            }
            Some(protobuf::partitioning::PartitionMethod::Unknown(partition_count)) => {
                Ok(Some(Partitioning::UnknownPartitioning(
                    *partition_count as usize,
                )))
            }
            None => Ok(None),
        },
        None => Ok(None),
    }
}

pub fn parse_protobuf_file_scan_schema(
    proto: &protobuf::FileScanExecConf,
) -> Result<Arc<Schema>> {
    Ok(Arc::new(convert_required!(proto.schema)?))
}

/// Parses a TableSchema from protobuf, extracting the file schema and partition columns
pub fn parse_table_schema_from_proto(
    proto: &protobuf::FileScanExecConf,
) -> Result<TableSchema> {
    let schema: Arc<Schema> = parse_protobuf_file_scan_schema(proto)?;

    // Reacquire the partition column types from the schema before removing them below.
    let table_partition_cols = proto
        .table_partition_cols
        .iter()
        .map(|col| Ok(Arc::new(schema.field_with_name(col)?.clone())))
        .collect::<Result<Vec<_>>>()?;

    // Remove partition columns from the schema after recreating table_partition_cols
    // because the partition columns are not in the file. They are present to allow
    // the partition column types to be reconstructed after serde.
    let file_schema = Arc::new(
        Schema::new(
            schema
                .fields()
                .iter()
                .filter(|field| !table_partition_cols.contains(field))
                .cloned()
                .collect::<Vec<_>>(),
        )
        .with_metadata(schema.metadata.clone()),
    );

    Ok(TableSchema::new(file_schema, table_partition_cols))
}

pub fn parse_protobuf_file_scan_config(
    proto: &protobuf::FileScanExecConf,
    ctx: &PhysicalPlanDecodeContext<'_>,
    proto_converter: &dyn PhysicalProtoConverterExtension,
    file_source: Arc<dyn FileSource>,
) -> Result<FileScanConfig> {
    let schema: Arc<Schema> = parse_protobuf_file_scan_schema(proto)?;

    let constraints = convert_required!(proto.constraints)?;
    let statistics = convert_required!(proto.statistics)?;

    let file_groups = proto
        .file_groups
        .iter()
        .map(|f| f.try_into())
        .collect::<Result<Vec<_>, _>>()?;

    let object_store_url = match proto.object_store_url.is_empty() {
        false => ObjectStoreUrl::parse(&proto.object_store_url)?,
        true => ObjectStoreUrl::local_filesystem(),
    };

    let mut output_ordering = vec![];
    for node_collection in &proto.output_ordering {
        let sort_exprs = parse_physical_sort_exprs(
            &node_collection.physical_sort_expr_nodes,
            ctx,
            &schema,
            proto_converter,
        )?;
        output_ordering.extend(LexOrdering::new(sort_exprs));
    }

    // Parse projection expressions if present and apply to file source
    let file_source = if let Some(proto_projection_exprs) = &proto.projection_exprs {
        let projection_exprs: Vec<ProjectionExpr> = proto_projection_exprs
            .projections
            .iter()
            .map(|proto_expr| {
                let expr = proto_converter.proto_to_physical_expr(
                    proto_expr.expr.as_ref().ok_or_else(|| {
                        internal_datafusion_err!("ProjectionExpr missing expr field")
                    })?,
                    &schema,
                    ctx,
                )?;
                Ok(ProjectionExpr::new(expr, proto_expr.alias.clone()))
            })
            .collect::<Result<Vec<_>>>()?;

        let projection_exprs = ProjectionExprs::new(projection_exprs);

        // Apply projection to file source
        file_source
            .try_pushdown_projection(&projection_exprs)?
            .unwrap_or(file_source)
    } else {
        file_source
    };

    let config = FileScanConfigBuilder::new(object_store_url, file_source)
        .with_file_groups(file_groups)
        .with_constraints(constraints)
        .with_statistics(statistics)
        .with_limit(proto.limit.as_ref().map(|sl| sl.limit as usize))
        .with_output_ordering(output_ordering)
        .with_batch_size(proto.batch_size.map(|s| s as usize))
        .build();
    Ok(config)
}

pub fn parse_record_batches(buf: &[u8]) -> Result<Vec<RecordBatch>> {
    if buf.is_empty() {
        return Ok(vec![]);
    }
    let reader = StreamReader::try_new(buf, None)?;
    let mut batches = Vec::new();
    for batch in reader {
        batches.push(batch?);
    }
    Ok(batches)
}

impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
    type Error = DataFusionError;

    fn try_from(val: &protobuf::PartitionedFile) -> Result<Self, Self::Error> {
        let mut pf = PartitionedFile::new_from_meta(ObjectMeta {
            location: Path::parse(val.path.as_str())
                .map_err(|e| proto_error(format!("Invalid object_store path: {e}")))?,
            last_modified: Utc.timestamp_nanos(val.last_modified_ns as i64),
            size: val.size,
            e_tag: None,
            version: None,
        })
        .with_partition_values(
            val.partition_values
                .iter()
                .map(|v| v.try_into())
                .collect::<Result<Vec<_>, _>>()?,
        );
        if let Some(range) = val.range.as_ref() {
            let file_range: FileRange = range.try_into()?;
            pf = pf.with_range(file_range.start, file_range.end);
        }
        if let Some(proto_stats) = val.statistics.as_ref() {
            pf = pf.with_statistics(Arc::new(proto_stats.try_into()?));
        }
        Ok(pf)
    }
}

impl TryFrom<&protobuf::FileRange> for FileRange {
    type Error = DataFusionError;

    fn try_from(value: &protobuf::FileRange) -> Result<Self, Self::Error> {
        Ok(FileRange {
            start: value.start,
            end: value.end,
        })
    }
}

impl TryFrom<&protobuf::FileGroup> for FileGroup {
    type Error = DataFusionError;

    fn try_from(val: &protobuf::FileGroup) -> Result<Self, Self::Error> {
        let files = val
            .files
            .iter()
            .map(|f| f.try_into())
            .collect::<Result<Vec<_>, _>>()?;
        Ok(FileGroup::new(files))
    }
}

impl TryFrom<&protobuf::JsonSink> for JsonSink {
    type Error = DataFusionError;

    fn try_from(value: &protobuf::JsonSink) -> Result<Self, Self::Error> {
        Ok(Self::new(
            convert_required!(value.config)?,
            convert_required!(value.writer_options)?,
        ))
    }
}

#[cfg(feature = "parquet")]
impl TryFrom<&protobuf::ParquetSink> for ParquetSink {
    type Error = DataFusionError;

    fn try_from(value: &protobuf::ParquetSink) -> Result<Self, Self::Error> {
        Ok(Self::new(
            convert_required!(value.config)?,
            convert_required!(value.parquet_options)?,
        ))
    }
}

impl TryFrom<&protobuf::CsvSink> for CsvSink {
    type Error = DataFusionError;

    fn try_from(value: &protobuf::CsvSink) -> Result<Self, Self::Error> {
        Ok(Self::new(
            convert_required!(value.config)?,
            convert_required!(value.writer_options)?,
        ))
    }
}

impl TryFrom<&protobuf::FileSinkConfig> for FileSinkConfig {
    type Error = DataFusionError;

    fn try_from(conf: &protobuf::FileSinkConfig) -> Result<Self, Self::Error> {
        let file_group = FileGroup::new(
            conf.file_groups
                .iter()
                .map(|f| f.try_into())
                .collect::<Result<Vec<_>>>()?,
        );
        let table_paths = conf
            .table_paths
            .iter()
            .map(ListingTableUrl::parse)
            .collect::<Result<Vec<_>>>()?;
        let table_partition_cols = conf
            .table_partition_cols
            .iter()
            .map(|protobuf::PartitionColumn { name, arrow_type }| {
                let data_type = convert_required!(arrow_type)?;
                Ok((name.clone(), data_type))
            })
            .collect::<Result<Vec<_>>>()?;
        let insert_op = match conf.insert_op() {
            protobuf::InsertOp::Append => InsertOp::Append,
            protobuf::InsertOp::Overwrite => InsertOp::Overwrite,
            protobuf::InsertOp::Replace => InsertOp::Replace,
        };
        let file_output_mode = match conf.file_output_mode() {
            protobuf::FileOutputMode::Automatic => {
                datafusion_datasource::file_sink_config::FileOutputMode::Automatic
            }
            protobuf::FileOutputMode::SingleFile => {
                datafusion_datasource::file_sink_config::FileOutputMode::SingleFile
            }
            protobuf::FileOutputMode::Directory => {
                datafusion_datasource::file_sink_config::FileOutputMode::Directory
            }
        };
        Ok(Self {
            original_url: String::default(),
            object_store_url: ObjectStoreUrl::parse(&conf.object_store_url)?,
            file_group,
            table_paths,
            output_schema: Arc::new(convert_required!(conf.output_schema)?),
            table_partition_cols,
            insert_op,
            keep_partition_by_columns: conf.keep_partition_by_columns,
            file_extension: conf.file_extension.clone(),
            file_output_mode,
        })
    }
}

#[cfg(test)]
mod tests {

    use super::*;

    #[test]
    fn partitioned_file_path_roundtrip_percent_encoded() {
        let path_str = "foo/foo%2Fbar/baz%252Fqux";
        let pf = PartitionedFile::new_from_meta(ObjectMeta {
            location: Path::parse(path_str).unwrap(),
            last_modified: Utc.timestamp_nanos(1_000),
            size: 42,
            e_tag: None,
            version: None,
        });

        let proto = protobuf::PartitionedFile::try_from(&pf).unwrap();
        assert_eq!(proto.path, path_str);

        let pf2 = PartitionedFile::try_from(&proto).unwrap();
        assert_eq!(pf2.object_meta.location.as_ref(), path_str);
        assert_eq!(pf2.object_meta.location, pf.object_meta.location);
        assert_eq!(pf2.object_meta.size, pf.object_meta.size);
        assert_eq!(pf2.object_meta.last_modified, pf.object_meta.last_modified);
    }

    #[test]
    fn partitioned_file_from_proto_invalid_path() {
        let proto = protobuf::PartitionedFile {
            path: "foo//bar".to_string(),
            size: 1,
            last_modified_ns: 0,
            partition_values: vec![],
            range: None,
            statistics: None,
        };

        let err = PartitionedFile::try_from(&proto).unwrap_err();
        assert!(err.to_string().contains("Invalid object_store path"));
    }
}