Function test_ffi_session

datafusion/ffi/src/session/mod.rs:659–721

()

    #[tokio::test]
    async fn test_ffi_session() -> Result<(), DataFusionError> {
        let (ctx, task_ctx_provider) = crate::util::tests::test_session_and_ctx();
        let mut table_options = TableOptions::default();
        table_options.csv.has_header = Some(true);
        table_options.json.schema_infer_max_rec = Some(10);
        table_options.parquet.global.coerce_int96 = Some("123456789".into());
        table_options.current_format = Some(ConfigFileType::JSON);

        let state = SessionStateBuilder::new_from_existing(ctx.state())
            .with_table_options(table_options)
            .build();

        let logical_codec = FFI_LogicalExtensionCodec::new(
            Arc::new(DefaultLogicalExtensionCodec {}),
            None,
            task_ctx_provider,
        );

        let local_session = FFI_SessionRef::new(&state, None, logical_codec);
        let foreign_session = ForeignSession::try_from(&local_session)?;

        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
        let df_schema = schema.try_into()?;
        let physical_expr = foreign_session.create_physical_expr(col("a"), &df_schema)?;
        assert_eq!(
            format!("{physical_expr:?}"),
            "Column { name: \"a\", index: 0 }"
        );

        assert_eq!(foreign_session.session_id(), state.session_id());

        let logical_plan = LogicalPlan::default();
        let physical_plan = foreign_session.create_physical_plan(&logical_plan).await?;
        assert_eq!(
            format!("{physical_plan:?}"),
            "EmptyExec { schema: Schema { fields: [], metadata: {} }, partitions: 1, cache: PlanProperties { eq_properties: EquivalenceProperties { eq_group: EquivalenceGroup { map: {}, classes: [] }, oeq_class: OrderingEquivalenceClass { orderings: [] }, oeq_cache: OrderingEquivalenceCache { normal_cls: OrderingEquivalenceClass { orderings: [] }, leading_map: {} }, constraints: Constraints { inner: [] }, schema: Schema { fields: [], metadata: {} } }, partitioning: UnknownPartitioning(1), emission_type: Incremental, boundedness: Bounded, evaluation_type: Lazy, scheduling_type: Cooperative, output_ordering: None } }"
        );

        assert_eq!(
            format!("{:?}", foreign_session.default_table_options()),
            format!("{:?}", state.default_table_options())
        );

        assert_eq!(
            format!("{:?}", foreign_session.table_options()),
            format!("{:?}", state.table_options())
        );

        let local_udfs = state.udfs();
        for udf in foreign_session.scalar_functions().keys() {
            assert!(local_udfs.contains(udf));
        }
        let local_udafs = state.udafs();
        for udaf in foreign_session.aggregate_functions().keys() {
            assert!(local_udafs.contains(udaf));
        }
        let local_udwfs = state.udwfs();
        for udwf in foreign_session.window_functions().keys() {