Function test_sql_insert_into_fifo

datafusion/core/tests/fifo/mod.rs:329–410

()

    /// It tests the INSERT INTO functionality.
    #[tokio::test]
    async fn test_sql_insert_into_fifo() -> Result<()> {
        // To make unbounded deterministic
        let waiting = Arc::new(AtomicBool::new(true));
        let waiting_thread = waiting.clone();
        // create local execution context
        let config = SessionConfig::new().with_batch_size(TEST_BATCH_SIZE);
        let ctx = SessionContext::new_with_config(config);
        // Create a new temporary FIFO file
        let tmp_dir = TempDir::new()?;
        let source_fifo_path = create_fifo_file(&tmp_dir, "source.csv")?;
        // Prevent move
        let (source_fifo_path_thread, source_display_fifo_path) =
            (source_fifo_path.clone(), source_fifo_path.display());
        // Tasks
        let mut tasks: Vec<JoinHandle<()>> = vec![];
        tasks.push(create_writing_task(
            source_fifo_path_thread,
            "a1,a2\n".to_owned(),
            (0..TEST_DATA_SIZE)
                .map(|_| "a,1\n".to_string())
                .collect::<Vec<_>>(),
            waiting,
            SEND_BEFORE_WAITING,
        ));
        // Create a new temporary FIFO file
        let sink_fifo_path = create_fifo_file(&tmp_dir, "sink.csv")?;
        // Prevent move
        let (sink_fifo_path_thread, sink_display_fifo_path) =
            (sink_fifo_path.clone(), sink_fifo_path.display());

        // Spawn a new thread to read sink EXTERNAL TABLE.
        #[expect(clippy::disallowed_methods)] // spawn allowed only in tests
        tasks.push(spawn_blocking(move || {
            let file = File::open(sink_fifo_path_thread).unwrap();
            let schema = Arc::new(Schema::new(vec![
                Field::new("a1", DataType::Utf8, false),
                Field::new("a2", DataType::UInt32, false),
            ]));

            let mut reader = ReaderBuilder::new(schema)
                .with_batch_size(TEST_BATCH_SIZE)
                .with_header(true)
                .build(file)
                .unwrap();

            while let Some(Ok(_)) = reader.next() {
                waiting_thread.store(false, Ordering::SeqCst);
            }
        }));
        // register second csv file with the SQL (create an empty file if not found)
        ctx.sql(&format!(
            "CREATE UNBOUNDED EXTERNAL TABLE source_table (
                a1  VARCHAR NOT NULL,
                a2  INT NOT NULL
            )
            STORED AS CSV
            LOCATION '{source_display_fifo_path}'
            OPTIONS ('format.has_header' 'true')"