Fix out of order row groups in filter query (#1974)

- Add test for ordered filter query
- Configure `datafusion` to prefer existing sort

This ensures queries with filter clauses return data in order.

apache/datafusion#10572 (comment)

Author: twitu

nautilus_core/persistence/src/backend/session.rs

@@ -71,8 +71,9 @@ impl DataBackendSession {
             .enable_all()
             .build()
             .unwrap();
-        let session_cfg =
-            SessionConfig::new().set_str("datafusion.optimizer.repartition_file_scans", "false");
+        let session_cfg = SessionConfig::new()
+            .set_str("datafusion.optimizer.repartition_file_scans", "false")
+            .set_str("datafusion.optimizer.prefer_existing_sort", "true");
         let session_ctx = SessionContext::new_with_config(session_cfg);
         Self {
             session_ctx,
@@ -119,7 +120,7 @@ impl DataBackendSession {
             file_sort_order: vec![vec![Expr::Sort(Sort {
                 expr: Box::new(col("ts_init")),
                 asc: true,
-                nulls_first: true,
+                nulls_first: false,
             })]],
             ..Default::default()
         };
@@ -129,7 +130,7 @@ impl DataBackendSession {
             parquet_options,
         ))?;
 
-        let default_query = format!("SELECT * FROM {}", &table_name);
+        let default_query = format!("SELECT * FROM {} ORDER BY ts_init", &table_name);
         let sql_query = sql_query.unwrap_or(&default_query);
         let query = self.runtime.block_on(self.session_ctx.sql(sql_query))?;
 

nautilus_core/persistence/tests/test_catalog.rs

@@ -237,6 +237,22 @@ fn test_quote_tick_query() {
     assert!(is_monotonically_increasing_by_init(&ticks));
 }
 
+#[rstest]
+fn test_quote_tick_query_with_filter() {
+    let file_path = "../../tests/test_data/nautilus/quotes-3-groups-filter-query.parquet";
+    let mut catalog = DataBackendSession::new(10);
+    catalog
+        .add_file::<QuoteTick>(
+            "quote_005",
+            file_path,
+            Some("SELECT * FROM quote_005 WHERE ts_init >= 1701388832486000000 ORDER BY ts_init"),
+        )
+        .unwrap();
+    let query_result: QueryResult = catalog.get_query_result();
+    let ticks: Vec<Data> = query_result.collect();
+    assert!(is_monotonically_increasing_by_init(&ticks));
+}
+
 #[rstest]
 fn test_quote_tick_multiple_query() {
     let expected_length = 9_600;

nautilus_trader/persistence/catalog/parquet.py

@@ -622,6 +622,7 @@ def _build_query(
         if conditions:
             query += f" WHERE {' AND '.join(conditions)}"
 
+        query += " ORDER BY ts_init"
         return query
 
     @staticmethod