@@ -26,6 +26,7 @@
 use std::rc::Rc;
 use std::sync::Arc;
 
+use datafusion::arrow::compute::kernels::cast_utils::parse_interval_month_day_nano;
 use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field};
 use datafusion::catalog_common::MemorySchemaProvider;
 use datafusion::common::config::ConfigOptions;
@@ -40,6 +41,7 @@ use datafusion::logical_expr::function::{PartitionEvaluatorArgs, WindowUDFFieldA
 use datafusion::logical_expr::{
     self, lit, AggregateUDF, Case, Distinct, Expr, JoinType, LogicalPlan, PartitionEvaluator,
     ScalarUDF, ScalarUDFImpl, Signature, SimpleAggregateUDF, TypeSignature, Volatility,
+    WindowFrameBound as DfWindowFrameBound, WindowFrameUnits as DfWindowFrameUnits,
     WindowFunctionDefinition, WindowUDF, WindowUDFImpl,
 };
 use datafusion::optimizer::analyzer::function_rewrite::ApplyFunctionRewrites;
@@ -50,7 +52,8 @@ use asap_sql_function_catalog::{AggSemantic, Arity, RewriteKind};
 use asap_types::pre_asap::agg_intent::AggIntent;
 use asap_types::pre_asap::query_expr::{
     GroupKeys, Predicate, ProjectItem, Reduction, SortKey, Source,
-    UnresolvedQueryExpr as Unresolved,
+    UnresolvedQueryExpr as Unresolved, WindowFrame, WindowFrameBound, WindowFrameOffset,
+    WindowFrameUnits,
 };
 use asap_types::pre_asap::schema::{DataType, Schema};
 use asap_types::pre_asap::{
@@ -67,7 +70,7 @@ mod types;
 pub use types::SqlCatalog;
 
 use self::expr::df_expr_to_unresolved;
-use self::types::{arrow_to_dtype, schema_to_arrow};
+use self::types::{arrow_to_dtype, scalar_value_to_asap, schema_to_arrow};
 
 std::thread_local! {
     static ACCURACY: std::cell::RefCell<AccuracyTarget> =
@@ -557,8 +560,8 @@ impl<'a> SqlLowerer<'a> {
         })
     }
 
-    /// `func(args) OVER (PARTITION BY … ORDER BY …)`. One window function per
-    /// plan node; window frames are not modelled yet (default frame assumed).
+    /// `func(args) OVER (PARTITION BY … ORDER BY … ROWS/RANGE BETWEEN …)`. One
+    /// window function per plan node.
     fn lower_window(&self, window: &logical_expr::Window) -> Result<Unresolved, LoweringError> {
         if window.window_expr.len() > 1 {
             return Err(LoweringError::UnsupportedFeature(format!(
@@ -613,6 +616,7 @@ impl<'a> SqlLowerer<'a> {
                 })
             })
             .collect::<Result<Vec<_>, _>>()?;
+        let frame = lower_window_frame(&wf.window_frame)?;
         // The window plan's schema is `[input fields …, window output]`; the last
         // field is the window column's name (what an enclosing Project references).
         let output_name = window
@@ -626,6 +630,7 @@ impl<'a> SqlLowerer<'a> {
             args,
             partition_by: partition_by.into(),
             order_by,
+            frame: Some(frame),
             output_name,
             child,
         })
@@ -1742,6 +1747,79 @@ fn lower_window_func_kind(fun: &WindowFunctionDefinition) -> Result<WindowFuncKi
     }
 }
 
+/// Map DataFusion's resolved `WindowFrame` (issue #268) to the canonical
+/// [`WindowFrame`]. DataFusion's planner always fills in the SQL-standard
+/// default frame before the logical plan is built, so this never sees an
+/// "absent" frame — only `ROWS`/`RANGE`/`GROUPS` with concrete bounds.
+/// `GROUPS` is rejected: no query in this repo's SQL corpora uses it, and
+/// nothing downstream interprets frame semantics yet, so it isn't worth
+/// modelling untested.
+fn lower_window_frame(
+    frame: &datafusion::logical_expr::WindowFrame,
+) -> Result<WindowFrame, LoweringError> {
+    let units = match frame.units {
+        DfWindowFrameUnits::Rows => WindowFrameUnits::Rows,
+        DfWindowFrameUnits::Range => WindowFrameUnits::Range,
+        DfWindowFrameUnits::Groups => {
+            return Err(LoweringError::UnsupportedFeature(
+                "window frame unit: GROUPS".into(),
+            ))
+        }
+    };
+    let offset = |v: &DfScalarValue| -> Result<WindowFrameOffset, LoweringError> {
+        Ok(match v {
+            DfScalarValue::IntervalYearMonth(Some(months)) => WindowFrameOffset::Interval {
+                months: *months,
+                days: 0,
+                nanoseconds: 0,
+            },
+            DfScalarValue::IntervalDayTime(Some(value)) => WindowFrameOffset::Interval {
+                months: 0,
+                days: value.days,
+                nanoseconds: i64::from(value.milliseconds) * 1_000_000,
+            },
+            DfScalarValue::IntervalMonthDayNano(Some(value)) => WindowFrameOffset::Interval {
+                months: value.months,
+                days: value.days,
+                nanoseconds: value.nanoseconds,
+            },
+            // DataFusion 43 keeps every RANGE offset as text: both numeric
+            // bounds such as `1.5` and normalized interval literals such as
+            // `"1 HOUR"`. Arrow's interval parser accepts bare numbers and
+            // interprets them as months, so classify numeric text first.
+            DfScalarValue::Utf8(Some(value)) | DfScalarValue::LargeUtf8(Some(value)) => {
+                if let Ok(value) = value.parse::<i64>() {
+                    WindowFrameOffset::Scalar(ScalarValue::Int64(value))
+                } else if let Ok(value) = value.parse::<f64>() {
+                    WindowFrameOffset::Scalar(ScalarValue::Float64(value))
+                } else {
+                    match parse_interval_month_day_nano(value) {
+                        Ok(interval) => WindowFrameOffset::Interval {
+                            months: interval.months,
+                            days: interval.days,
+                            nanoseconds: interval.nanoseconds,
+                        },
+                        Err(_) => WindowFrameOffset::Scalar(scalar_value_to_asap(v)?),
+                    }
+                }
+            }
+            _ => WindowFrameOffset::Scalar(scalar_value_to_asap(v)?),
+        })
+    };
+    let bound = |b: &DfWindowFrameBound| -> Result<WindowFrameBound, LoweringError> {
+        Ok(match b {
+            DfWindowFrameBound::Preceding(v) => WindowFrameBound::Preceding(offset(v)?),
+            DfWindowFrameBound::CurrentRow => WindowFrameBound::CurrentRow,
+            DfWindowFrameBound::Following(v) => WindowFrameBound::Following(offset(v)?),
+        })
+    };
+    Ok(WindowFrame {
+        units,
+        start_bound: bound(&frame.start_bound)?,
+        end_bound: bound(&frame.end_bound)?,
+    })
+}
+
 // ── Issue #225, item 3: DataFusion registry drift detection ────────────────
 //
 // `asap_sql_function_catalog::NATIVE_FUNCTIONS` is hand-maintained data
 
@@ -9,7 +9,7 @@ use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog};
 use asap_types::pre_asap::schema::{Column, DataType, Schema};
 use asap_types::pre_asap::{
     AggIntent, CompareOpKind, GroupKeys, JoinKind, QueryExpr, Reduction, ScalarValue, Source,
-    WindowFuncKind,
+    WindowFrameBound, WindowFrameOffset, WindowFrameUnits, WindowFuncKind,
 };
 use asap_types::types::AccuracyTarget;
 use asap_types::workload::SqlDialect;
@@ -722,6 +722,136 @@ async fn window_aggregate_lowers_to_windowfunc() {
     assert_eq!(args, &vec![QueryExpr::Column(3)], "SUM(bytes) → arg col 3");
 }
 
+// ── Window frames (issue #268) ───────────────────────────────────────────────
+
+/// The frame clause must actually reach the IR, not just the display string:
+/// three window frames that differ semantically must lower to different
+/// `SQLWindowFunc.frame` values.
+#[tokio::test]
+async fn window_frame_is_captured_not_dropped() {
+    let default_frame =
+        lower("SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts) FROM metrics")
+            .await;
+    let two_preceding = lower(
+        "SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts \
+         ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) FROM metrics",
+    )
+    .await;
+    let unbounded_following = lower(
+        "SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts \
+         ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) FROM metrics",
+    )
+    .await;
+
+    let frame_of = |qe: &QueryExpr| {
+        let QueryExpr::SQLWindowFunc { frame, .. } = find_windowfunc(qe).unwrap() else {
+            unreachable!();
+        };
+        frame
+            .clone()
+            .expect("newly lowered SQL always records a frame")
+    };
+    let (a, b, c) = (
+        frame_of(&default_frame),
+        frame_of(&two_preceding),
+        frame_of(&unbounded_following),
+    );
+    assert_ne!(a, b, "default frame vs ROWS 2 PRECEDING must differ");
+    assert_ne!(
+        a, c,
+        "default frame vs ROWS CURRENT..UNBOUNDED FOLLOWING must differ"
+    );
+    assert_ne!(b, c);
+
+    assert_eq!(b.units, WindowFrameUnits::Rows);
+    assert_eq!(
+        b.start_bound,
+        WindowFrameBound::Preceding(WindowFrameOffset::Scalar(ScalarValue::Int64(2)))
+    );
+    assert_eq!(b.end_bound, WindowFrameBound::CurrentRow);
+
+    assert_eq!(c.start_bound, WindowFrameBound::CurrentRow);
+    assert_eq!(
+        c.end_bound,
+        WindowFrameBound::Following(WindowFrameOffset::Scalar(ScalarValue::Null))
+    );
+}
+
+#[tokio::test]
+async fn range_interval_frame_is_preserved() {
+    let qe = lower(
+        "SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts \
+         RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW) FROM metrics",
+    )
+    .await;
+    let QueryExpr::SQLWindowFunc {
+        frame: Some(frame), ..
+    } = find_windowfunc(&qe).unwrap()
+    else {
+        panic!("expected a window function with a concrete frame");
+    };
+
+    assert_eq!(frame.units, WindowFrameUnits::Range);
+    assert_eq!(
+        frame.start_bound,
+        WindowFrameBound::Preceding(WindowFrameOffset::Interval {
+            months: 0,
+            days: 0,
+            nanoseconds: 3_600_000_000_000,
+        })
+    );
+    assert_eq!(frame.end_bound, WindowFrameBound::CurrentRow);
+}
+
+#[tokio::test]
+async fn range_numeric_frames_remain_scalar_offsets() {
+    let integer = lower(
+        "SELECT SUM(bytes) OVER (ORDER BY bytes \
+         RANGE BETWEEN 2 PRECEDING AND CURRENT ROW) FROM metrics",
+    )
+    .await;
+    let fractional = lower(
+        "SELECT SUM(latency) OVER (ORDER BY latency \
+         RANGE BETWEEN 1.5 PRECEDING AND CURRENT ROW) FROM metrics",
+    )
+    .await;
+
+    let start_bound = |qe: &QueryExpr| {
+        let QueryExpr::SQLWindowFunc {
+            frame: Some(frame), ..
+        } = find_windowfunc(qe).unwrap()
+        else {
+            panic!("expected a window function with a concrete frame");
+        };
+        frame.start_bound.clone()
+    };
+
+    assert_eq!(
+        start_bound(&integer),
+        WindowFrameBound::Preceding(WindowFrameOffset::Scalar(ScalarValue::Int64(2)))
+    );
+    assert_eq!(
+        start_bound(&fractional),
+        WindowFrameBound::Preceding(WindowFrameOffset::Scalar(ScalarValue::Float64(1.5)))
+    );
+}
+
+/// `GROUPS` frames aren't in this repo's SQL corpora and nothing downstream
+/// interprets frame semantics yet — rejected explicitly rather than silently
+/// mis-lowered.
+#[tokio::test]
+async fn groups_frame_is_rejected() {
+    let err = lower_sql(
+        "SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts \
+         GROUPS BETWEEN 2 PRECEDING AND CURRENT ROW) FROM metrics",
+        &catalog(),
+        AccuracyTarget::Exact,
+    )
+    .await
+    .expect_err("GROUPS frame unit must be rejected");
+    assert!(format!("{err}").contains("GROUPS"), "got {err}");
+}
+
 // ── Nested query functions: derived tables / inline views (issue #27) ───────────
 
 /// Collect every `AggIntent` in the tree, root-to-leaf.
 
@@ -499,6 +499,7 @@ fn build(expr: &QueryExpr, nodes: &mut Vec<DagNode>, cache: &mut HashCache) -> u
             args,
             partition_by,
             order_by,
+            frame,
             output_name,
             child,
         } => {
@@ -508,6 +509,7 @@ fn build(expr: &QueryExpr, nodes: &mut Vec<DagNode>, cache: &mut HashCache) -> u
                 "args": args,
                 "partition_by": partition_by,
                 "order_by": order_by,
+                "frame": frame,
                 "output_name": output_name,
             });
             push_node(
 
@@ -294,7 +294,10 @@ fn try_rewrite_rownumber_topk(expr: &QueryExpr) -> Option<QueryExpr> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::pre_asap::query_expr::{GroupKeys, ProjectItem, Source};
+    use crate::pre_asap::query_expr::{
+        GroupKeys, ProjectItem, Source, WindowFrame, WindowFrameBound, WindowFrameOffset,
+        WindowFrameUnits,
+    };
     use crate::pre_asap::schema::{Column, DataType, Schema};
     use crate::types::AccuracyTarget;
 
@@ -473,6 +476,16 @@ mod tests {
         }
     }
 
+    /// `ROW_NUMBER` ignores its frame clause, so the top-k rewrite doesn't care
+    /// what's in it; any concrete frame works as fixture data.
+    fn rownumber_frame() -> WindowFrame {
+        WindowFrame {
+            units: WindowFrameUnits::Rows,
+            start_bound: WindowFrameBound::Preceding(WindowFrameOffset::Scalar(ScalarValue::Null)),
+            end_bound: WindowFrameBound::Following(WindowFrameOffset::Scalar(ScalarValue::Null)),
+        }
+    }
+
     /// `Filter{ rn(3) <= 5 } { SQLWindowFunc{ RowNumber, PARTITION BY region(2),
     /// ORDER BY col(2) DESC } { agg } }`.
     fn rownumber_topk(agg: QueryExpr) -> QueryExpr {
@@ -485,6 +498,7 @@ mod tests {
                 ascending: false,
                 nulls_first: true,
             }],
+            frame: Some(rownumber_frame()),
             output_name: "rn".into(),
             child: Rc::new(agg),
         };
@@ -567,6 +581,7 @@ mod tests {
                 ascending: false,
                 nulls_first: true,
             }],
+            frame: Some(rownumber_frame()),
             output_name: "rn".into(),
             child: Rc::new(grouped(AggIntent::Count {
                 accuracy: AccuracyTarget::Exact,
 
@@ -365,6 +365,7 @@ pub fn structural_hash(node: &QueryExpr, cache: &mut HashCache) -> u64 {
             args,
             partition_by,
             order_by,
+            frame,
             output_name,
             child,
         } => {
@@ -376,6 +377,7 @@ pub fn structural_hash(node: &QueryExpr, cache: &mut HashCache) -> u64 {
                     args,
                     partition_by,
                     order_by,
+                    frame,
                     output_name,
                 ),
             );
@@ -682,13 +684,15 @@ fn rebuild_children(table: &mut InternTable, expr: QueryExpr) -> QueryExpr {
             args,
             partition_by,
             order_by,
+            frame,
             output_name,
             child,
         } => SQLWindowFunc {
             func,
             args,
             partition_by,
             order_by,
+            frame,
             output_name,
             child: intern_child(table, child),
         },
 
@@ -56,7 +56,8 @@ pub use query_expr::{
     aggregate_output_schema, AtModifier, BinaryOpKind, ColState, DataModel, GroupKeys, GroupSide,
     InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError, Reduction,
     ResolvedQueryExpr, SampleKind, SetOpKind, SortKey, Source, TimeShift, UnresolvedQueryExpr,
-    VectorGrouping, VectorMatch, VectorMatchKind, WindowFuncKind,
+    VectorGrouping, VectorMatch, VectorMatchKind, WindowFrame, WindowFrameBound, WindowFrameOffset,
+    WindowFrameUnits, WindowFuncKind,
 };
 pub use resolve::{resolve_root, ResolveTreeError};
 pub use schema::{Column, ColumnId, DataType, Schema};