[adapters] Don't materialize tables with PK+lateness.

ryzhyk · ryzhyk · commit 806b7a56da1e · 2025-10-26T18:06:23.000Z
Signed-off-by: Leonid Ryzhyk &lt;ryzhyk@gmail.com&gt;
diff --git a/crates/adapters/src/static_compile/catalog.rs b/crates/adapters/src/static_compile/catalog.rs
@@ -259,7 +259,7 @@ impl Catalog {
         .unwrap();
 
         // Inputs are also outputs.
-        self.register_materialized_output_map_persistent(
+        self.register_output_map_persistent(
             Self::output_persistent_id(&stream).as_deref(),
             stream,
             schema,
@@ -506,6 +506,33 @@ impl Catalog {
         self.register_materialized_output_map_persistent(None, stream, schema)
     }
 
+    pub fn register_output_map_persistent<K, KD, V, VD>(
+        &mut self,
+        persistent_id: Option<&str>,
+        stream: Stream<RootCircuit, OrdIndexedZSet<K, V>>,
+        schema: &str,
+    ) where
+        KD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig>
+            + SerializeWithContext<SqlSerdeConfig>
+            + From<K>
+            + Send
+            + Sync
+            + Debug
+            + 'static,
+        VD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig>
+            + SerializeWithContext<SqlSerdeConfig>
+            + From<V>
+            + Default
+            + Debug
+            + Clone
+            + Send
+            + 'static,
+        K: DBData + Send + Sync + From<KD> + Default,
+        V: DBData + Send + Sync + From<VD> + Default,
+    {
+        self.register_output_map_persistent_inner(persistent_id, stream, schema, false)
+    }
+
     pub fn register_materialized_output_map_persistent<K, KD, V, VD>(
         &mut self,
         persistent_id: Option<&str>,
@@ -529,6 +556,34 @@ impl Catalog {
             + 'static,
         K: DBData + Send + Sync + From<KD> + Default,
         V: DBData + Send + Sync + From<VD> + Default,
+    {
+        self.register_output_map_persistent_inner(persistent_id, stream, schema, true)
+    }
+
+    pub fn register_output_map_persistent_inner<K, KD, V, VD>(
+        &mut self,
+        persistent_id: Option<&str>,
+        stream: Stream<RootCircuit, OrdIndexedZSet<K, V>>,
+        schema: &str,
+        materialized: bool,
+    ) where
+        KD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig>
+            + SerializeWithContext<SqlSerdeConfig>
+            + From<K>
+            + Send
+            + Sync
+            + Debug
+            + 'static,
+        VD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig>
+            + SerializeWithContext<SqlSerdeConfig>
+            + From<V>
+            + Default
+            + Debug
+            + Clone
+            + Send
+            + 'static,
+        K: DBData + Send + Sync + From<KD> + Default,
+        V: DBData + Send + Sync + From<VD> + Default,
     {
         let schema: Relation = Self::parse_relation_schema(schema).unwrap();
         let name = schema.name.clone();
@@ -546,16 +601,26 @@ impl Catalog {
         let (delta_handle, delta_gid) = delta.accumulate_output_persistent_with_gid(persistent_id);
         stream.circuit().set_mir_node_id(&delta_gid, persistent_id);
 
-        // `integrate_trace` below should return the existing integral created by the InputUpsert operator.
-        let (integrate_handle, integral_gid) = stream
-            .integrate_trace()
-            .apply(|s| TypedBatch::<K, V, ZWeight, _>::new(s.inner().ro_snapshot()))
-            .output_persistent_with_gid(
-                persistent_id.map(|id| format!("{id}.integral")).as_deref(),
-            );
-        stream
-            .circuit()
-            .set_mir_node_id(&integral_gid, persistent_id);
+        let integrate_handle = if materialized {
+            // `integrate_trace` below should return the existing integral created by the InputUpsert operator.
+            let (integrate_handle, integral_gid) = stream
+                .integrate_trace()
+                .apply(|s| TypedBatch::<K, V, ZWeight, _>::new(s.inner().ro_snapshot()))
+                .output_persistent_with_gid(
+                    persistent_id
+                        .map(|id| format!("{id}.output_integral"))
+                        .as_deref(),
+                );
+            stream
+                .circuit()
+                .set_mir_node_id(&integral_gid, persistent_id);
+            Some(
+                Arc::new(<SerCollectionHandleImpl<_, KD, VD>>::new(integrate_handle))
+                    as Arc<dyn SerBatchReaderHandle>,
+            )
+        } else {
+            None
+        };
 
         let handles = OutputCollectionHandles {
             key_schema: None,
@@ -564,9 +629,7 @@ impl Catalog {
             delta_handle: Box::new(<SerCollectionHandleImpl<_, VD, ()>>::new(delta_handle))
                 as Box<dyn SerBatchReaderHandle>,
             integrate_handle_is_indexed: true,
-            integrate_handle: Some(Arc::new(<SerCollectionHandleImpl<_, KD, VD>>::new(
-                integrate_handle,
-            )) as Arc<dyn SerBatchReaderHandle>),
+            integrate_handle,
         };
 
         self.register_output_batch_handles(&name, handles).unwrap();
diff --git a/docs.feldera.com/docs/sql/materialized.md b/docs.feldera.com/docs/sql/materialized.md
@@ -19,9 +19,10 @@ To declare a materialized table, use the materialized attribute:
 CREATE TABLE my_table (...) WITH ('materialized' = 'true');
 ```
 
-In addition, tables with a `PRIMARY KEY` constraint are automatically materialized
-(since the pipeline must in any case maintain an index of such a table used to enforce the constraint)
-and do not require an explicit `materialized` attribute.
+Tables with a `PRIMARY KEY` constraint are automatically materialized, since the pipeline
+must maintain an index to enforce the constraint. Therefore, such tables do not need an explicit
+`materialized` attribute. The only exception is when a table has a `LATENESS` attribute defined
+on one of its primary key columns—in that case, you must specify the `materialized` attribute explicitly.
 
 To declare a materialized view, use the `CREATE MATERIALIZED VIEW` syntax:
 
