Add refresh_options to ImportOpSpec.

Author: badmonster0

src/base/spec.rs

@@ -1,8 +1,7 @@
-use std::ops::Deref;
-
-use serde::{Deserialize, Serialize};
+use crate::prelude::*;
 
 use super::schema::{EnrichedValueType, FieldSchema};
+use std::ops::Deref;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "kind")]
@@ -163,9 +162,17 @@ pub struct OpSpec {
     pub spec: serde_json::Map<String, serde_json::Value>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct SourceRefreshOptions {
+    pub refresh_interval: Option<chrono::Duration>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ImportOpSpec {
     pub source: OpSpec,
+
+    #[serde(default)]
+    pub refresh_options: SourceRefreshOptions,
 }
 
 /// Transform data using a given operator.

src/builder/analyzer.rs

@@ -668,6 +668,7 @@ impl AnalyzerContext<'_> {
                     .typ
                     .clone(),
                 name: op_name,
+                refresh_options: import_op.spec.refresh_options,
             })
         };
         Ok(result_fut)

src/builder/flow_builder.rs

@@ -380,13 +380,14 @@ impl FlowBuilder {
         self.root_data_scope_ref.clone()
     }
 
-    #[pyo3(signature = (kind, op_spec, target_scope, name))]
+    #[pyo3(signature = (kind, op_spec, target_scope, name, refresh_options=None))]
     pub fn add_source(
         &mut self,
         kind: String,
         op_spec: py::Pythonized<serde_json::Map<String, serde_json::Value>>,
         target_scope: Option<DataScopeRef>,
         name: String,
+        refresh_options: Option<py::Pythonized<spec::SourceRefreshOptions>>,
     ) -> PyResult<DataSlice> {
         if let Some(target_scope) = target_scope {
             if !Arc::ptr_eq(&target_scope.0, &self.root_data_scope_ref.0) {
@@ -402,6 +403,7 @@ impl FlowBuilder {
                     kind,
                     spec: op_spec.into_inner(),
                 },
+                refresh_options: refresh_options.map(|o| o.into_inner()).unwrap_or_default(),
             },
         };
         let analyzer_ctx = AnalyzerContext {

src/builder/plan.rs

@@ -1,9 +1,5 @@
-use std::sync::Arc;
+use crate::prelude::*;
 
-use serde::Serialize;
-
-use crate::base::schema::ValueType;
-use crate::base::value;
 use crate::execution::db_tracking_setup;
 use crate::ops::interface::*;
 use crate::utils::fingerprint::{Fingerprint, Fingerprinter};
@@ -60,7 +56,8 @@ pub struct AnalyzedImportOp {
     pub source_id: i32,
     pub executor: Box<dyn SourceExecutor>,
     pub output: AnalyzedOpOutput,
-    pub primary_key_type: ValueType,
+    pub primary_key_type: schema::ValueType,
+    pub refresh_options: spec::SourceRefreshOptions,
 }
 
 pub struct AnalyzedFunctionExecInfo {
@@ -69,7 +66,7 @@ pub struct AnalyzedFunctionExecInfo {
 
     /// Fingerprinter of the function's behavior.
     pub fingerprinter: Fingerprinter,
-    pub output_type: ValueType,
+    pub output_type: schema::ValueType,
 }
 
 pub struct AnalyzedTransformOp {
@@ -106,7 +103,7 @@ pub struct AnalyzedExportOp {
     pub executor: Arc<dyn ExportTargetExecutor>,
     pub query_target: Option<Arc<dyn QueryTarget>>,
     pub primary_key_def: AnalyzedPrimaryKeyDef,
-    pub primary_key_type: ValueType,
+    pub primary_key_type: schema::ValueType,
     /// idx for value fields - excluding the primary key field.
     pub value_fields: Vec<u32>,
     /// If true, value is never changed on the same primary key.