readyset: Add --parsing-preset CLI option

We plumb it through to everywhere that was previously using a hardcoded
`Default::default()`. It can also be set with the `PARSING_PRESET`
environment variable.

Addresses: REA-5047
Release-Note-Core: Add a `--parsing-preset` CLI option and
  `PARSING_PRESET` environment variable which can be used to configure
  how we handle mismatches between nom-sql and sqlparser-rs parsing. It
  defaults to `prefer-nom` which parses with both nom-sql and
  sqlparser-rs, logs a warning if they don't agree, and uses the AST
  produced by nom-sql. To disable parsing with sqlparser-rs, use
  `only-nom`. To test experimental query support for syntax not
  supported by nom, use `prefer-sqlparser` or `only-sqlparser`.

Change-Id: I6a6a6964f5168268cc6983c8e8df79c64e7ed36a
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9683
Reviewed-by: Mohamed Yasser Abdelhamed Abdeen <mohamed@readyset.io>
Tested-by: Buildkite CI

Author: mvzink

Cargo.lock

@@ -280,6 +280,7 @@ pub struct BackendBuilder {
     client_addr: SocketAddr,
     slowlog: bool,
     dialect: Dialect,
+    parsing_preset: ParsingPreset,
     users: HashMap<String, String>,
     require_authentication: bool,
     query_log_sender: Option<UnboundedSender<QueryExecutionEvent>>,
@@ -301,6 +302,7 @@ impl Default for BackendBuilder {
             client_addr: SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 0)),
             slowlog: false,
             dialect: Dialect::MySQL,
+            parsing_preset: ParsingPreset::for_prod(),
             users: Default::default(),
             require_authentication: true,
             query_log_sender: None,
@@ -363,6 +365,7 @@ impl BackendBuilder {
             settings: BackendSettings {
                 slowlog: self.slowlog,
                 dialect: self.dialect,
+                parsing_preset: self.parsing_preset,
                 require_authentication: self.require_authentication,
                 unsupported_set_mode: self.unsupported_set_mode,
                 migration_mode: self.migration_mode,
@@ -396,6 +399,11 @@ impl BackendBuilder {
         self
     }
 
+    pub fn parsing_preset(mut self, parsing_preset: ParsingPreset) -> Self {
+        self.parsing_preset = parsing_preset;
+        self
+    }
+
     pub fn query_log_sender(
         mut self,
         query_log_sender: Option<UnboundedSender<QueryExecutionEvent>>,
@@ -609,6 +617,8 @@ where
 struct BackendSettings {
     /// SQL dialect to use when parsing queries from clients
     dialect: Dialect,
+    /// Parsing mode that determines which parser(s) to use and how to handle conflicts
+    parsing_preset: ParsingPreset,
     slowlog: bool,
     require_authentication: bool,
     /// How to behave when receiving unsupported `SET` statements
@@ -3191,7 +3201,7 @@ where
     fn parse_query(&mut self, query: &str) -> ReadySetResult<SqlQuery> {
         trace!(%query, "Parsing query");
         match readyset_sql_parsing::parse_query_with_config(
-            ParsingPreset::for_prod(),
+            self.settings.parsing_preset,
             self.settings.dialect,
             query,
         ) {

readyset-adapter/src/backend.rs

@@ -43,7 +43,7 @@ use readyset_sql::ast::{
     CreateTableStatement, CreateViewStatement, DropTableStatement, DropViewStatement,
     NonReplicatedRelation, Relation, SelectStatement, SqlIdentifier, SqlQuery, TableKey,
 };
-use readyset_sql_parsing::{parse_query_with_config, ParsingPreset};
+use readyset_sql_parsing::{parse_query_with_config, ParsingConfig, ParsingPreset};
 use readyset_sql_passes::adapter_rewrites::{self, AdapterRewriteParams};
 use serde::{Deserialize, Serialize};
 use test_strategy::Arbitrary;
@@ -138,23 +138,27 @@ impl ChangeList {
         }
     }
 
-    /// Parse a `ChangeList` from the given vector of SQL strings, using the given [`Dialect`]
-    /// for expression evaluation semantics.
+    /// Parse a `ChangeList` from the given vector of SQL strings, using the given [`Dialect`] for
+    /// expression evaluation semantics (but it will be parsed with the canonical MySQL dialect).
     ///
     /// This method processes each string in the vector individually and constructs a list of
     /// changes. If any query fails to parse, the method returns an error.
     pub fn from_strings(queries: Vec<impl AsRef<str>>, dialect: Dialect) -> ReadySetResult<Self> {
-        let mut changes = Vec::new();
+        Self::from_strings_with_config(queries, dialect, ParsingPreset::for_prod().into_config())
+    }
 
+    pub fn from_strings_with_config(
+        queries: Vec<impl AsRef<str>>,
+        dialect: Dialect,
+        parsing_config: ParsingConfig,
+    ) -> ReadySetResult<Self> {
+        let mut changes = Vec::new();
         for query_str in queries {
-            let parsed = parse_query_with_config(
-                ParsingPreset::BothPreferNom,
-                Dialect::DEFAULT_MYSQL.into(),
-                &query_str,
-            )
-            .map_err(|_| ReadySetError::UnparseableQuery {
-                query: query_str.as_ref().to_string(),
-            })?;
+            let parsed =
+                parse_query_with_config(parsing_config, Dialect::DEFAULT_MYSQL.into(), &query_str)
+                    .map_err(|_| ReadySetError::UnparseableQuery {
+                        query: query_str.as_ref().to_string(),
+                    })?;
 
             match parsed {
                 SqlQuery::CreateTable(statement) => changes.push(Change::CreateTable {
@@ -449,6 +453,7 @@ impl Change {
     pub fn from_cache_ddl_request(
         ddl_req: &CacheDDLRequest,
         adapter_rewrite_params: AdapterRewriteParams,
+        parsing_preset: ParsingPreset,
     ) -> ReadySetResult<Self> {
         macro_rules! mk_error {
             ($str:expr) => {
@@ -458,7 +463,7 @@ impl Change {
             };
         }
         match parse_query_with_config(
-            ParsingPreset::for_prod(),
+            parsing_preset.into_config().log_on_mismatch(true),
             ddl_req.dialect.into(),
             &ddl_req.unparsed_stmt,
         ) {

readyset-client/src/recipe/changelist.rs

@@ -10,6 +10,7 @@ use readyset_client::consensus::{
     Authority, LocalAuthority, LocalAuthorityStore, NodeTypeSchedulingRestriction,
     WorkerSchedulingConfig,
 };
+use readyset_sql_parsing::ParsingPreset;
 use readyset_telemetry_reporter::TelemetrySender;
 use readyset_util::shutdown::{self, ShutdownSender};
 
@@ -27,6 +28,9 @@ pub struct Builder {
     external_addr: SocketAddr,
     leader_eligible: bool,
     domain_scheduling_config: WorkerSchedulingConfig,
+    /// The parsing preset to use for SQL parsing. Public so we can use it when setting up replication
+    /// separately in tests.
+    pub parsing_preset: ParsingPreset,
     /// The telemetry sender
     pub telemetry: TelemetrySender,
     wait_for_failpoint: bool,
@@ -44,6 +48,7 @@ impl Default for Builder {
             memory_check_frequency: None,
             leader_eligible: true,
             domain_scheduling_config: Default::default(),
+            parsing_preset: ParsingPreset::for_prod(),
             telemetry: TelemetrySender::new_no_op(),
             wait_for_failpoint: false,
             unquery: true,
@@ -131,6 +136,7 @@ impl Builder {
         builder.set_abort_on_task_failure(false);
         builder.set_full_materialization(true);
         builder.set_post_lookup(true);
+        builder.set_parsing_preset(ParsingPreset::BothPanicOnMismatch);
         builder
     }
 
@@ -374,6 +380,10 @@ impl Builder {
         self.config.replicator_statement_logging = value;
     }
 
+    pub fn set_parsing_preset(&mut self, value: ParsingPreset) {
+        self.parsing_preset = value;
+    }
+
     /// Start a server instance and return a handle to it. This method also returns a
     /// [`ShutdownSender`] that should be used to shut down the server when it is no longer needed.
     pub fn start(
@@ -387,6 +397,7 @@ impl Builder {
             memory_limit,
             memory_check_frequency,
             domain_scheduling_config,
+            parsing_preset,
             leader_eligible,
             telemetry,
             wait_for_failpoint,
@@ -403,6 +414,7 @@ impl Builder {
             memory_limit,
             memory_check_frequency,
             domain_scheduling_config,
+            parsing_preset,
             leader_eligible,
             telemetry,
             wait_for_failpoint,
@@ -426,6 +438,7 @@ impl Builder {
             memory_limit,
             memory_check_frequency,
             domain_scheduling_config,
+            parsing_preset,
             leader_eligible,
             telemetry,
             wait_for_failpoint,
@@ -443,6 +456,7 @@ impl Builder {
             memory_limit,
             memory_check_frequency,
             domain_scheduling_config,
+            parsing_preset,
             leader_eligible,
             readers,
             reader_addr,

readyset-server/src/builder.rs

@@ -30,6 +30,7 @@ use readyset_client::{GraphvizOptions, ViewCreateRequest, WorkerDescriptor};
 use readyset_errors::{internal_err, ReadySetError, ReadySetResult};
 use readyset_sql::ast::Relation;
 use readyset_sql::Dialect;
+use readyset_sql_parsing::ParsingPreset;
 use readyset_telemetry_reporter::TelemetrySender;
 #[cfg(feature = "failure_injection")]
 use readyset_util::failpoints;
@@ -94,6 +95,9 @@ pub struct Leader {
     /// A client to the current authority.
     pub(super) authority: Arc<Authority>,
 
+    /// Parsing mode for the controller (used in extend_recipe and the replicator)
+    parsing_preset: ParsingPreset,
+
     /// A map of currently running migrations.
     ///
     /// Requests to `/extend_recipe` run for at least [`EXTEND_RECIPE_MAX_SYNC_TIME`], after which
@@ -206,6 +210,7 @@ impl Leader {
         let replicator_restart_timeout = self.replicator_config.replicator_restart_timeout;
         let config = self.replicator_config.clone();
         let replicator_statement_logging = self.replicator_statement_logging;
+        let parsing_preset = self.parsing_preset;
 
         // The replication task ideally won't panic, but if it does and we arent replicating, that
         // will mean the data we return, will be more and more stale, and the transaction logs on
@@ -230,6 +235,7 @@ impl Leader {
                         telemetry_sender.clone(),
                         server_startup,
                         replicator_statement_logging,
+                        parsing_preset,
                     )
                     .await
                     {
@@ -1050,6 +1056,7 @@ impl Leader {
         replicator_config: UpstreamConfig,
         worker_request_timeout: Duration,
         background_recovery_interval: Duration,
+        parsing_mode: ParsingPreset,
         replicator_tx: UnboundedSender<ReplicatorMessage>,
         controller_tx: UnboundedSender<ControllerMessage>,
     ) -> Self {
@@ -1069,6 +1076,7 @@ impl Leader {
             worker_request_timeout,
             background_recovery_interval,
             background_recovery_running: Arc::new(AtomicBool::new(false)),
+            parsing_preset: parsing_mode,
             running_migrations: Default::default(),
             background_task_failed,
             running_recovery: None,

readyset-server/src/controller/inner.rs

@@ -23,6 +23,7 @@ use readyset_client::ControllerDescriptor;
 use readyset_data::Dialect;
 use readyset_errors::{internal, internal_err, ReadySetError, ReadySetResult};
 use readyset_sql::ast::Relation;
+use readyset_sql_parsing::ParsingPreset;
 use readyset_telemetry_reporter::TelemetrySender;
 #[cfg(feature = "failure_injection")]
 use readyset_util::failpoints;
@@ -377,6 +378,10 @@ pub struct Controller {
     /// Whether we are the leader and ready to handle requests.
     leader_ready: Arc<AtomicBool>,
 
+    /// Parsing mode for the controller, used in extend_recipe. Not part of the serialized
+    /// [`Config`] so it can be changed on restart.
+    parsing_preset: ParsingPreset,
+
     /// Whether we are in maintenance mode
     maintenance_mode: Arc<AtomicBool>,
 
@@ -409,6 +414,7 @@ impl Controller {
         worker_descriptor: WorkerDescriptor,
         telemetry_sender: TelemetrySender,
         config: Config,
+        parsing_preset: ParsingPreset,
         shutdown_rx: ShutdownReceiver,
     ) -> Self {
         // If we don't have an upstream, we allow permissive writes to base tables.
@@ -425,6 +431,7 @@ impl Controller {
             our_descriptor,
             worker_descriptor,
             config,
+            parsing_preset,
             leader_ready: Arc::new(AtomicBool::new(false)),
             maintenance_mode: Arc::new(AtomicBool::new(false)),
             controller_channel: ControllerChannel::new(),
@@ -534,6 +541,7 @@ impl Controller {
                     self.config.replicator_config.clone(),
                     self.config.worker_request_timeout,
                     self.config.background_recovery_interval,
+                    self.parsing_preset,
                     self.replicator_channel.sender(),
                     self.controller_channel.sender(),
                 );
@@ -855,7 +863,11 @@ impl Controller {
             ddl_reqs
                 .iter()
                 .filter(|req| matches!(
-                    Change::from_cache_ddl_request(req, adapter_rewrite_params),
+                    Change::from_cache_ddl_request(
+                        req,
+                        adapter_rewrite_params,
+                        self.parsing_preset,
+                    ),
                     Ok(Change::Drop { .. })
                 ))
                 .all(|req| req.dialect == ddl_reqs[0].dialect),
@@ -866,7 +878,11 @@ impl Controller {
             ddl_reqs
                 .iter()
                 .filter(|req| matches!(
-                    Change::from_cache_ddl_request(req, adapter_rewrite_params),
+                    Change::from_cache_ddl_request(
+                        req,
+                        adapter_rewrite_params,
+                        self.parsing_preset,
+                    ),
                     Ok(Change::Drop { .. })
                 ))
                 .all(|req| req.schema_search_path.is_empty()),
@@ -877,7 +893,11 @@ impl Controller {
             ddl_reqs
                 .iter()
                 .filter(|req| matches!(
-                    Change::from_cache_ddl_request(req, adapter_rewrite_params),
+                    Change::from_cache_ddl_request(
+                        req,
+                        adapter_rewrite_params,
+                        self.parsing_preset,
+                    ),
                     Ok(Change::CreateCache(_))
                 ))
                 .all(|req| req.dialect == ddl_reqs[0].dialect),
@@ -891,7 +911,11 @@ impl Controller {
         let mut last_was_drop = false;
 
         for ddl_req in ddl_reqs {
-            match Change::from_cache_ddl_request(&ddl_req, adapter_rewrite_params) {
+            match Change::from_cache_ddl_request(
+                &ddl_req,
+                adapter_rewrite_params,
+                self.parsing_preset,
+            ) {
                 Ok(change) => {
                     let is_drop = matches!(change, Change::Drop { .. });
 

readyset-server/src/controller/mod.rs

@@ -16,6 +16,7 @@ use readyset_server::metrics::{
 };
 use readyset_server::PrometheusBuilder;
 use readyset_server::{resolve_addr, Builder, NoriaMetricsRecorder, WorkerOptions};
+use readyset_sql_parsing::ParsingPreset;
 use readyset_telemetry_reporter::{TelemetryEvent, TelemetryInitializer};
 use readyset_version::*;
 use tracing::{error, info};
@@ -147,6 +148,17 @@ struct Options {
     /// impact startup.
     #[arg(long, hide = true)]
     wait_for_failpoint: bool,
+
+    /// Parsing mode that determines which parser(s) to use and how to handle conflicts.
+    /// Options: only-nom, only-sqlparser, prefer-nom, prefer-sqlparser, panic
+    #[arg(
+        long,
+        env = "PARSING_PRESET",
+        value_enum,
+        default_value = "both-prefer-nom",
+        hide = true
+    )]
+    parsing_preset: ParsingPreset,
 }
 
 fn main() -> anyhow::Result<()> {
@@ -247,6 +259,7 @@ fn main() -> anyhow::Result<()> {
     builder.set_telemetry_sender(telemetry_sender.clone());
     builder.set_wait_for_failpoint(opts.wait_for_failpoint);
     builder.set_replicator_statement_logging(opts.tracing.statement_logging);
+    builder.set_parsing_preset(opts.parsing_preset);
 
     if opts.cannot_become_leader {
         builder.cannot_become_leader();