From f4654682609133fb75c32df5bf5019bf1de8f59a Mon Sep 17 00:00:00 2001
From: Arnaud de Bossoreille <arnaud.debossoreille@gmail.com>
Date: Mon, 10 Feb 2025 20:52:51 +0100
Subject: [PATCH] Fix flat data assertion in CSV writer

---
 contrib/quirky_binder_csv/src/write.rs | 19 +++++++++++++------
 1 file changed, 13 insertions(+), 6 deletions(-)

diff --git a/contrib/quirky_binder_csv/src/write.rs b/contrib/quirky_binder_csv/src/write.rs
index 22d5eae..b802ccf 100644
--- a/contrib/quirky_binder_csv/src/write.rs
+++ b/contrib/quirky_binder_csv/src/write.rs
@@ -25,16 +25,23 @@ pub struct WriteCsv {
 
 impl WriteCsv {
     fn new<R: TypeResolver + Copy>(
-        _graph: &mut GraphBuilder<R>,
+        graph: &mut GraphBuilder<R>,
         name: FullyQualifiedName,
         inputs: [NodeStream; 1],
         params: WriteCsvParams,
     ) -> ChainResultWithTrace<Self> {
-        if !inputs.single().sub_streams().is_empty() {
-            return Err(ChainError::Other {
-                msg: "Sub streams are not supported".to_owned(),
-            })
-            .with_trace_element(trace_element!(WRITE_CSV_TRACE_NAME));
+        let input = inputs.single();
+        let input_stream_def = graph
+            .get_stream(input.record_type())
+            .with_trace_element(trace_element!(WRITE_CSV_TRACE_NAME))?
+            .borrow();
+        for d in input_stream_def.get_current_data() {
+            if input.sub_streams().contains_key(&d) {
+                return Err(ChainError::Other {
+                    msg: "Sub streams are not supported".to_owned(),
+                })
+                .with_trace_element(trace_element!(WRITE_CSV_TRACE_NAME));
+            }
         }
 
         Ok(Self {