Change tracking arg name to track_rows_processed

Author: treysp

sqlmesh/core/console.py

@@ -4306,7 +4306,7 @@ def _calculate_annotation_str_len(
 # Convert number of bytes to a human-readable string
 # https://github.com/dbt-labs/dbt-adapters/blob/34fd178539dcb6f82e18e738adc03de7784c032f/dbt-bigquery/src/dbt/adapters/bigquery/connections.py#L165
 def _format_bytes(num_bytes: t.Optional[int]) -> str:
-    if num_bytes and num_bytes > 0:
+    if num_bytes and num_bytes >= 0:
         if num_bytes < 1024:
             return f"{num_bytes} bytes"
 
@@ -4324,7 +4324,7 @@ def _format_bytes(num_bytes: t.Optional[int]) -> str:
 # Abbreviate integer count. Example: 1,000,000,000 -> 1b
 # https://github.com/dbt-labs/dbt-adapters/blob/34fd178539dcb6f82e18e738adc03de7784c032f/dbt-bigquery/src/dbt/adapters/bigquery/connections.py#L178
 def _abbreviate_integer_count(count: t.Optional[int]) -> str:
-    if count and count > 0:
+    if count and count >= 0:
         if count < 1000:
             return str(count)
 

sqlmesh/core/engine_adapter/base.py

@@ -845,7 +845,7 @@ def _create_table_from_source_queries(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         table = exp.to_table(table_name)
@@ -891,15 +891,15 @@ def _create_table_from_source_queries(
                             replace=replace,
                             table_description=table_description,
                             table_kind=table_kind,
-                            track_execution_stats=track_execution_stats,
+                            track_rows_processed=track_rows_processed,
                             **kwargs,
                         )
                     else:
                         self._insert_append_query(
                             table_name,
                             query,
                             target_columns_to_types or self.columns(table),
-                            track_execution_stats=track_execution_stats,
+                            track_rows_processed=track_rows_processed,
                         )
 
         # Register comments with commands if the engine supports comments and we weren't able to
@@ -923,7 +923,7 @@ def _create_table(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         self.execute(
@@ -941,7 +941,7 @@ def _create_table(
                 table_kind=table_kind,
                 **kwargs,
             ),
-            track_execution_stats=track_execution_stats,
+            track_rows_processed=track_rows_processed,
         )
 
     def _build_create_table_exp(
@@ -1429,7 +1429,7 @@ def insert_append(
         table_name: TableName,
         query_or_df: QueryOrDF,
         target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         source_columns: t.Optional[t.List[str]] = None,
     ) -> None:
         source_queries, target_columns_to_types = self._get_source_queries_and_columns_to_types(
@@ -1439,15 +1439,15 @@ def insert_append(
             source_columns=source_columns,
         )
         self._insert_append_source_queries(
-            table_name, source_queries, target_columns_to_types, track_execution_stats
+            table_name, source_queries, target_columns_to_types, track_rows_processed
         )
 
     def _insert_append_source_queries(
         self,
         table_name: TableName,
         source_queries: t.List[SourceQuery],
         target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
     ) -> None:
         with self.transaction(condition=len(source_queries) > 0):
             target_columns_to_types = target_columns_to_types or self.columns(table_name)
@@ -1457,7 +1457,7 @@ def _insert_append_source_queries(
                         table_name,
                         query,
                         target_columns_to_types,
-                        track_execution_stats=track_execution_stats,
+                        track_rows_processed=track_rows_processed,
                     )
 
     def _insert_append_query(
@@ -1466,13 +1466,13 @@ def _insert_append_query(
         query: Query,
         target_columns_to_types: t.Dict[str, exp.DataType],
         order_projections: bool = True,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
     ) -> None:
         if order_projections:
             query = self._order_projections_and_filter(query, target_columns_to_types)
         self.execute(
             exp.insert(query, table_name, columns=list(target_columns_to_types)),
-            track_execution_stats=track_execution_stats,
+            track_rows_processed=track_rows_processed,
         )
 
     def insert_overwrite_by_partition(
@@ -1615,7 +1615,7 @@ def _insert_overwrite_by_condition(
                         )
                         if insert_overwrite_strategy.is_replace_where:
                             insert_exp.set("where", where or exp.true())
-                        self.execute(insert_exp, track_execution_stats=True)
+                        self.execute(insert_exp, track_rows_processed=True)
 
     def update_table(
         self,
@@ -1637,7 +1637,7 @@ def _merge(
             exp.Subquery(this=query), alias=MERGE_SOURCE_ALIAS, copy=False, table=True
         )
         self.execute(
-            exp.Merge(this=this, using=using, on=on, whens=whens), track_execution_stats=True
+            exp.Merge(this=this, using=using, on=on, whens=whens), track_rows_processed=True
         )
 
     def scd_type_2_by_time(
@@ -2387,7 +2387,7 @@ def execute(
         expressions: t.Union[str, exp.Expression, t.Sequence[exp.Expression]],
         ignore_unsupported_errors: bool = False,
         quote_identifiers: bool = True,
-        track_execution_stats: bool = False,
+        track_rows_processed: bool = False,
         **kwargs: t.Any,
     ) -> None:
         """Execute a sql query."""
@@ -2409,7 +2409,7 @@ def execute(
                     expression=e if isinstance(e, exp.Expression) else None,
                     quote_identifiers=quote_identifiers,
                 )
-                self._execute(sql, track_execution_stats, **kwargs)
+                self._execute(sql, track_rows_processed, **kwargs)
 
     def _attach_correlation_id(self, sql: str) -> str:
         if self.ATTACH_CORRELATION_ID and self.correlation_id:
@@ -2439,12 +2439,12 @@ def _record_execution_stats(
     ) -> None:
         QueryExecutionTracker.record_execution(sql, rowcount, bytes_processed)
 
-    def _execute(self, sql: str, track_execution_stats: bool = False, **kwargs: t.Any) -> None:
+    def _execute(self, sql: str, track_rows_processed: bool = False, **kwargs: t.Any) -> None:
         self.cursor.execute(sql, **kwargs)
 
         if (
             self.SUPPORTS_QUERY_EXECUTION_TRACKING
-            and track_execution_stats
+            and track_rows_processed
             and QueryExecutionTracker.is_tracking()
         ):
             rowcount_raw = getattr(self.cursor, "rowcount", None)
@@ -2501,7 +2501,7 @@ def temp_table(
                 exists=True,
                 table_description=None,
                 column_descriptions=None,
-                track_execution_stats=False,
+                track_rows_processed=False,
                 **kwargs,
             )
 
@@ -2753,7 +2753,7 @@ def _replace_by_key(
                     insert_statement.set("where", delete_filter)
                     insert_statement.set("this", exp.to_table(target_table))
 
-                self.execute(insert_statement, track_execution_stats=True)
+                self.execute(insert_statement, track_rows_processed=True)
             finally:
                 self.drop_table(temp_table)
 

sqlmesh/core/engine_adapter/bigquery.py

@@ -1052,7 +1052,7 @@ def _db_call(self, func: t.Callable[..., t.Any], *args: t.Any, **kwargs: t.Any)
     def _execute(
         self,
         sql: str,
-        track_execution_stats: bool = False,
+        track_rows_processed: bool = False,
         **kwargs: t.Any,
     ) -> None:
         """Execute a sql query."""
@@ -1098,7 +1098,7 @@ def _execute(
         self.cursor._set_rowcount(query_results)
         self.cursor._set_description(query_results.schema)
 
-        if track_execution_stats and QueryExecutionTracker.is_tracking():
+        if track_rows_processed and QueryExecutionTracker.is_tracking():
             num_rows = None
             if query_job.statement_type == "CREATE_TABLE_AS_SELECT":
                 # since table was just created, number rows in table == number rows processed

sqlmesh/core/engine_adapter/clickhouse.py

@@ -294,7 +294,7 @@ def _insert_overwrite_by_condition(
                     )
 
                 try:
-                    self.execute(existing_records_insert_exp, track_execution_stats=True)
+                    self.execute(existing_records_insert_exp, track_rows_processed=True)
                 finally:
                     if table_partition_exp:
                         self.drop_table(partitions_temp_table_name)
@@ -489,7 +489,7 @@ def _create_table(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         """Creates a table in the database.
@@ -526,7 +526,7 @@ def _create_table(
             column_descriptions,
             table_kind,
             empty_ctas=(self.engine_run_mode.is_cloud and expression is not None),
-            track_execution_stats=track_execution_stats,
+            track_rows_processed=track_rows_processed,
             **kwargs,
         )
 

sqlmesh/core/engine_adapter/duckdb.py

@@ -157,7 +157,7 @@ def _create_table(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         catalog = self.get_current_catalog()
@@ -181,7 +181,7 @@ def _create_table(
             table_description,
             column_descriptions,
             table_kind,
-            track_execution_stats=track_execution_stats,
+            track_rows_processed=track_rows_processed,
             **kwargs,
         )
 

sqlmesh/core/engine_adapter/redshift.py

@@ -174,7 +174,7 @@ def _create_table_from_source_queries(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         """
@@ -429,7 +429,7 @@ def resolve_target_table(expression: exp.Expression) -> exp.Expression:
                 on=on.transform(resolve_target_table),
                 whens=whens.transform(resolve_target_table),
             ),
-            track_execution_stats=True,
+            track_rows_processed=True,
         )
 
     def _normalize_decimal_value(self, expr: exp.Expression, precision: int) -> exp.Expression:

sqlmesh/core/engine_adapter/snowflake.py

@@ -170,7 +170,7 @@ def _create_table(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         table_format = kwargs.get("table_format")
@@ -190,7 +190,7 @@ def _create_table(
             table_description=table_description,
             column_descriptions=column_descriptions,
             table_kind=table_kind,
-            track_execution_stats=track_execution_stats,
+            track_rows_processed=track_rows_processed,
             **kwargs,
         )
 

sqlmesh/core/engine_adapter/spark.py

@@ -434,7 +434,7 @@ def _create_table(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         table_name = (
@@ -463,7 +463,7 @@ def _create_table(
             target_columns_to_types=target_columns_to_types,
             table_description=table_description,
             column_descriptions=column_descriptions,
-            track_execution_stats=track_execution_stats,
+            track_rows_processed=track_rows_processed,
             **kwargs,
         )
         table_name = (

sqlmesh/core/engine_adapter/trino.py

@@ -359,7 +359,7 @@ def _create_table(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         table_kind: t.Optional[str] = None,
-        track_execution_stats: bool = True,
+        track_rows_processed: bool = True,
         **kwargs: t.Any,
     ) -> None:
         super()._create_table(
@@ -371,7 +371,7 @@ def _create_table(
             table_description=table_description,
             column_descriptions=column_descriptions,
             table_kind=table_kind,
-            track_execution_stats=track_execution_stats,
+            track_rows_processed=track_rows_processed,
             **kwargs,
         )
 

sqlmesh/core/state_sync/db/environment.py

@@ -78,7 +78,7 @@ def update_environment(self, environment: Environment) -> None:
             self.environments_table,
             _environment_to_df(environment),
             target_columns_to_types=self._environment_columns_to_types,
-            track_execution_stats=False,
+            track_rows_processed=False,
         )
 
     def update_environment_statements(
@@ -109,7 +109,7 @@ def update_environment_statements(
                 self.environment_statements_table,
                 _environment_statements_to_df(environment_name, plan_id, environment_statements),
                 target_columns_to_types=self._environment_statements_columns_to_types,
-                track_execution_stats=False,
+                track_rows_processed=False,
             )
 
     def invalidate_environment(self, name: str, protect_prod: bool = True) -> None: