duckdb: honor preserve_insertion_order for scans

vortex-duckdb/cpp/multi_file_reader.cpp

@@ -84,10 +84,19 @@ VortexMultiFileReader::InitializeReader(MultiFileReaderData &reader_data,
 
     const VortexGlobalState &global = gstate.global_state->Cast<VortexGlobalState>();
 
+    // Unlike ANSI SQL, Duckdb provides ordering by default
+    // https://duckdb.org/docs/current/sql/dialect/order_preservation
+    Value ordered_value;
+    if (!context.TryGetCurrentSetting("preserve_insertion_order", ordered_value)) {
+        throw BinderException("preserve_insertion_order not set");
+    }
+    D_ASSERT(ordered_value.type() == LogicalType::BOOLEAN);
+    const bool ordered = ordered_value.GetValueUnsafe<bool>();
+
     duckdb_vx_error error = nullptr;
     const void *const ffi_global = global.ffi_global_state->DataPtr();
     void *const ffi_file = reader.ffi_file->DataPtr();
-    const bool skip = duckdb_reader_initialize(ffi_global, ffi_file, &error);
+    const bool skip = duckdb_reader_initialize(ffi_global, ffi_file, ordered, &error);
     if (error) {
         throw InvalidInputException(IntoErrString(error));
     }

vortex-duckdb/include/vortex.h

@@ -65,7 +65,11 @@ bool duckdb_reader_get_statistics(const void *file,
                                   size_t column_name_len,
                                   duckdb_column_statistics *stats_out);
 
-extern bool duckdb_reader_initialize(const void *global, void *file, duckdb_vx_error *error);
+extern
+bool duckdb_reader_initialize(const void *global,
+                              void *file,
+                              bool ordered,
+                              duckdb_vx_error *error);
 
 extern duckdb_logical_type duckdb_reader_bind_column_type(const void *bind, size_t index);
 

vortex-duckdb/src/ffi.rs

@@ -221,11 +221,12 @@ pub unsafe extern "C-unwind" fn duckdb_reader_get_statistics(
 pub unsafe extern "C-unwind" fn duckdb_reader_initialize(
     global: *const c_void,
     file: *mut c_void,
+    ordered: bool,
     error: *mut cpp::duckdb_vx_error,
 ) -> bool {
     let global = unsafe { global.cast::<GlobalState>().as_ref() }.vortex_expect("null pointer");
     let file = unsafe { file.cast::<OpenFileReader>().as_mut() }.vortex_expect("null pointer");
-    try_or(error, || reader_initialize(file, global))
+    try_or(error, || reader_initialize(file, global, ordered))
 }
 
 #[unsafe(no_mangle)]

vortex-duckdb/src/file_reader.rs

@@ -159,16 +159,21 @@ pub fn reader_bind(file: &OpenFileReader, result: &mut BindResultRef) -> VortexR
 /// Called once per file by one thread under file-local lock. Determines
 /// whether the opened file should be skipped. If this function returns false,
 /// duckdb closes the file and doesn't call reader_try_initialize_scan on it.
-pub fn reader_initialize(file: &mut OpenFileReader, global: &GlobalState) -> VortexResult<bool> {
+pub fn reader_initialize(
+    file: &mut OpenFileReader,
+    global: &GlobalState,
+    mut ordered: bool,
+) -> VortexResult<bool> {
     if file.can_skip(&global.filter)? {
         return Ok(true);
     }
 
-    // Getting splits is non-trivial work so we prefer doing it here under file
-    // lock and not in reader_try_initialize_scan under global lock.
-    let ordered = global.file_row_number_column_pos.is_some();
+    ordered |= global.file_row_number_column_pos.is_some();
+
     let reader = Arc::clone(&file.reader);
     let filter = &global.filter;
+    // Getting splits is non-trivial work so we prefer doing it here under file
+    // lock and not in reader_try_initialize_scan under global lock.
     let mut builder = ScanBuilder::new(SESSION.clone(), reader)
         .with_projection(global.projection.clone())
         .with_ordered(ordered)

vortex-sqllogictest/slt/duckdb/chunk-offset.slt

@@ -0,0 +1,27 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+include ../setup.slt.no
+
+statement ok
+COPY (SELECT range AS id FROM range(1000000))
+TO '${WORK_DIR}/chunk-offset.vortex';
+
+query I
+SELECT id FROM '${WORK_DIR}/chunk-offset.vortex' LIMIT 1;
+----
+0
+
+query I
+SELECT id FROM '${WORK_DIR}/chunk-offset.vortex'
+LIMIT 1 OFFSET 100000;
+----
+100000
+
+query I
+SELECT id FROM (
+    SELECT id, row_number() OVER () - 1 p
+    FROM '${WORK_DIR}/chunk-offset.vortex')
+WHERE p = 100000;
+----
+100000