GeezerVison

+/// A batch with reversed columns, an extra column, and nanosecond timestamps,
+/// read by position after `timestamps_to_micros` and `order_columns`, gives
+/// back the records that were written.
+#[test]
+fn delta_positional_read_matches_written_records() {
+    use std::sync::Arc;
+    use arrow::array::{ArrayRef, Int64Array, RecordBatch};
+    use arrow::compute::cast;
+    use arrow::datatypes::{DataType, Field as ArrowField, TimeUnit};
+    use feldera_adapterlib::utils::arrow_layout::{has_positional_fields, order_columns};
+    use feldera_adapterlib::utils::arrow_timestamps::timestamps_to_micros;
+    use feldera_types::transport::delta_table::DeltaVariantEncoding;
+
+    let records: Vec<DeltaTestStruct> = (0..50).map(|i| delta_test_record(i * 2)).collect();
+    let fields = DeltaTestStruct::schema();
+    let arrow_schema = ArrowSchema::new(relation_to_arrow_fields(&fields, delta_schema_options()));
+    let batch = serde_arrow::to_record_batch(arrow_schema.fields(),
+        &SerializeWithContextWrapper::new(&records, &delta_output_serde_config(DeltaVariantEncoding::Variant))).unwrap();
+    let mut new_fields = vec![ArrowField::new("extra", DataType::Int64, false)];
+    let mut new_columns: Vec<ArrayRef> = vec![Arc::new(Int64Array::from(vec![7; records.len()]))];
+    for (field, column) in batch.schema().fields().iter().zip(batch.columns()).rev() {
+        match field.data_type() {
+            DataType::Timestamp(_, tz) => {
+                let ns = DataType::Timestamp(TimeUnit::Nanosecond, tz.clone());
+                new_fields.push(field.as_ref().clone().with_data_type(ns.clone()));
+                new_columns.push(cast(column, &ns).unwrap());
+            }
+            _ => { new_fields.push(field.as_ref().clone()); new_columns.push(column.clone()); }
+        }
+    }
+    let batch = RecordBatch::try_new(Arc::new(ArrowSchema::new(new_fields)), new_columns).unwrap();
+    let batch = timestamps_to_micros(batch).unwrap();
+    let (batch, reason) = order_columns(batch, &fields);
+    assert_eq!(reason, None);
+    assert!(has_positional_fields(batch.schema_ref()));
+    let read: Vec<DeltaTestStruct> = Vec::deserialize_with_context(
+        serde_arrow::Deserializer::from_record_batch(&batch).unwrap(),
+        &delta_input_serde_config().with_positional_fields(true)).unwrap();
+    assert_eq!(read, records);
+}

diff --git a/crates/adapterlib/src/utils/arrow_layout.rs b/crates/adapterlib/src/utils/arrow_layout.rs
index 26fda65..da14dba 100644
--- a/crates/adapterlib/src/utils/arrow_layout.rs
+++ b/crates/adapterlib/src/utils/arrow_layout.rs
@@ -400,3 +400,37 @@ mod test {
         assert_eq!(ordered.column(1).data_type(), &DataType::Decimal128(10, 3));
     }
 }
+
+#[cfg(test)]
+mod qa_test {
+    use std::sync::Arc;
+
+    use arrow::array::{Array, ArrayRef, AsArray, Decimal128Array, RecordBatch};
+    use feldera_types::program_schema::{ColumnType, Field, SqlIdentifier};
+
+    use super::order_columns;
+
+    /// A sliced batch, as DataFusion produces, keeps the slice offset.
+    #[test]
+    fn sliced_decimal_significands() {
+        let decimals: ArrayRef = Arc::new(
+            Decimal128Array::from(vec![Some(1), Some(2), None, Some(4)])
+                .with_precision_and_scale(10, 2)
+                .unwrap(),
+        );
+        let batch = RecordBatch::try_from_iter(vec![("d", decimals)])
+            .unwrap()
+            .slice(1, 3);
+        let fields = [Field::new(
+            SqlIdentifier::new("d", false),
+            ColumnType::decimal(10, 2, true),
+        )];
+        let (ordered, reason) = order_columns(batch, &fields);
+        assert_eq!(reason, None);
+        let d = ordered.column(0).as_fixed_size_binary();
+        assert_eq!(d.len(), 3);
+        assert_eq!(d.value(0), 2_i128.to_ne_bytes());
+        assert!(d.is_null(1));
+        assert_eq!(d.value(2), 4_i128.to_ne_bytes());
+    }
+}
diff --git a/crates/static-connectors/src/test/delta_tests.rs b/crates/static-connectors/src/test/delta_tests.rs
index 77a05d0..3231ab4 100644
--- a/crates/static-connectors/src/test/delta_tests.rs
+++ b/crates/static-connectors/src/test/delta_tests.rs
@@ -264,3 +264,178 @@ fn delta_string_column_is_read_as_timestamp() {
     assert!(!positional);
     assert_eq!(read, expected);
 }
+
+#[derive(Debug, Default, PartialEq, Eq, Clone)]
+struct QaInner {
+    a: bool,
+    b: Option<i64>,
+}
+feldera_types::deserialize_table_record!(QaInner["QaInner", feldera_sqllib::SqlVariant, 2] {
+    (a, "a", false, bool, |_| None),
+    (b, "b", false, Option<i64>, |_| Some(None))
+});
+
+#[derive(Debug, Default, PartialEq, Eq, Clone)]
+struct QaOuter {
+    id: i64,
+    inner: Option<QaInner>,
+    list: Vec<QaInner>,
+}
+feldera_types::deserialize_table_record!(QaOuter["QaOuter", feldera_sqllib::SqlVariant, 3] {
+    (id, "id", false, i64, |_| None),
+    (inner, "inner", false, Option<QaInner>, |_| Some(None)),
+    (list, "list", false, Vec<QaInner>, |_| None)
+});
+
+/// A nullable nested ROW with reversed fields, generated as the compiler does,
+/// is read by position.
+#[test]
+fn qa_positional_nullable_nested_row() {
+    use arrow::array::{ArrayRef, BooleanArray, Int64Array, ListArray, RecordBatch, StructArray};
+    use arrow::buffer::{NullBuffer, OffsetBuffer};
+    use arrow::datatypes::{DataType, Field as AF, Fields};
+    use feldera_adapterlib::utils::arrow_layout::{has_positional_fields, order_columns};
+    use feldera_types::program_schema::{ColumnType, Field, SqlIdentifier};
+    use std::sync::Arc;
+
+    // Fields stored as (b, a).
+    let inner_fields = Fields::from(vec![
+        AF::new("b", DataType::Int64, true),
+        AF::new("A", DataType::Boolean, false),
+    ]);
+    let inner_cols = |b: Vec<Option<i64>>, a: Vec<bool>| -> Vec<ArrayRef> {
+        vec![Arc::new(Int64Array::from(b)), Arc::new(BooleanArray::from(a))]
+    };
+    let inner = StructArray::new(
+        inner_fields.clone(),
+        inner_cols(vec![Some(1), None, Some(3)], vec![true, false, true]),
+        Some(NullBuffer::from(vec![true, false, true])),
+    );
+    let items = StructArray::new(
+        inner_fields.clone(),
+        inner_cols(vec![Some(7), None], vec![false, true]),
+        None,
+    );
+    let list = ListArray::new(
+        Arc::new(AF::new("item", DataType::Struct(inner_fields), false)),
+        OffsetBuffer::from_lengths([2, 0, 0]),
+        Arc::new(items),
+        None,
+    );
+    let batch = RecordBatch::try_from_iter(vec![
+        ("list", Arc::new(list) as ArrayRef),
+        ("inner", Arc::new(inner) as ArrayRef),
+        ("id", Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef),
+    ])
+    .unwrap();
+
+    let f = |n: &str, t| Field::new(SqlIdentifier::new(n, false), t);
+    let row = ColumnType::structure(
+        true,
+        &[f("a", ColumnType::boolean(false)), f("b", ColumnType::bigint(true))],
+    );
+    let mut row_nn = row.clone();
+    row_nn.nullable = false;
+    let fields = [
+        f("id", ColumnType::bigint(false)),
+        f("inner", row),
+        f("list", ColumnType::array(false, row_nn)),
+    ];
+    let (batch, reason) = order_columns(batch, &fields);
+    assert_eq!(reason, None);
+    assert!(has_positional_fields(batch.schema_ref()));
+
+    let read: Vec<QaOuter> = Vec::deserialize_with_context(
+        serde_arrow::Deserializer::from_record_batch(&batch).unwrap(),
+        &delta_input_serde_config().with_positional_fields(true),
+    )
+    .unwrap();
+    assert_eq!(
+        read,
+        vec![
+            QaOuter {
+                id: 1,
+                inner: Some(QaInner { a: true, b: Some(1) }),
+                list: vec![
+                    QaInner { a: false, b: Some(7) },
+                    QaInner { a: true, b: None }
+                ],
+            },
+            QaOuter { id: 2, inner: None, list: vec![] },
+            QaOuter {
+                id: 3,
+                inner: Some(QaInner { a: true, b: Some(3) }),
+                list: vec![],
+            },
+        ]
+    );
+}
+
+#[derive(Debug, Default, PartialEq, Eq, Clone)]
+struct QaDec {
+    arr: Vec<Option<feldera_sqllib::SqlDecimal<10, 2>>>,
+    map: BTreeMap<String, feldera_sqllib::SqlDecimal<10, 2>>,
+}
+feldera_types::deserialize_table_record!(QaDec["QaDec", feldera_sqllib::SqlVariant, 2] {
+    (arr, "arr", false, Vec<Option<feldera_sqllib::SqlDecimal<10, 2>>>, |_| None),
+    (map, "map", false, BTreeMap<String, feldera_sqllib::SqlDecimal<10, 2>>, |_| None)
+});
+
+#[test]
+fn qa_positional_nested_decimals() {
+    use arrow::array::{
+        ArrayRef, Decimal128Array, ListArray, MapArray, RecordBatch, StringArray, StructArray,
+    };
+    use arrow::buffer::OffsetBuffer;
+    use arrow::datatypes::{DataType, Field as AF, Fields};
+    use feldera_adapterlib::utils::arrow_layout::order_columns;
+    use feldera_types::program_schema::{ColumnType, Field, SqlIdentifier};
+    use std::sync::Arc;
+
+    let dec = |v: Vec<Option<i128>>| {
+        Arc::new(Decimal128Array::from(v).with_precision_and_scale(10, 2).unwrap()) as ArrayRef
+    };
+    let arr = ListArray::new(
+        Arc::new(AF::new("item", DataType::Decimal128(10, 2), true)),
+        OffsetBuffer::from_lengths([2]),
+        dec(vec![Some(12345), None]),
+        None,
+    );
+    let entries = StructArray::new(
+        Fields::from(vec![
+            AF::new("key", DataType::Utf8, false),
+            AF::new("value", DataType::Decimal128(10, 2), false),
+        ]),
+        vec![Arc::new(StringArray::from(vec!["k"])) as ArrayRef, dec(vec![Some(-5)])],
+        None,
+    );
+    let map = MapArray::new(
+        Arc::new(AF::new("entries", arrow::array::Array::data_type(&entries).clone(), false)),
+        OffsetBuffer::from_lengths([1]),
+        entries,
+        None,
+        false,
+    );
+    let batch = RecordBatch::try_from_iter(vec![
+        ("map", Arc::new(map) as ArrayRef),
+        ("arr", Arc::new(arr) as ArrayRef),
+    ])
+    .unwrap();
+    let f = |n: &str, t| Field::new(SqlIdentifier::new(n, false), t);
+    let fields = [
+        f("arr", ColumnType::array(false, ColumnType::decimal(10, 2, true))),
+        f(
+            "map",
+            ColumnType::map(false, ColumnType::varchar(false), ColumnType::decimal(10, 2, false)),
+        ),
+    ];
+    let (batch, reason) = order_columns(batch, &fields);
+    assert_eq!(reason, None);
+    let read: Vec<QaDec> = Vec::deserialize_with_context(
+        serde_arrow::Deserializer::from_record_batch(&batch).unwrap(),
+        &delta_input_serde_config().with_positional_fields(true),
+    )
+    .unwrap();
+    assert_eq!(read[0].arr, vec![Some("123.45".parse().unwrap()), None]);
+    assert_eq!(read[0].map.get("k"), Some(&"-0.05".parse().unwrap()));
+}