feat: enriched staging query macros + unified import + source wrapping (#438)

## Summary

1. added offset and bound support to staging query macros `{{ start_date
}}` is valid as before, now `{{ start_date(offset=-10,
lower_bound='2023-01-01') }}` is also valid

2. Previously we required users to pass in quotes around the macro
separately. This pr removes the need for it
`{{ start_date }}` used to become `2023-01-01`, it now becomes
`'2023-01-01'`

2. added a unified top level module `api.chronon.types` that contain
everything that users need.

3. added wrappers on source sub types to directly return sources 
```py
ttypes.Source(events=ttypes.EventSource(...))

# now becomes
EventSource(...)
```

## Cheour clientslist
- [ ] Added Unit Tests
- [ ] Covered by existing CI
- [ ] Integration tested
- [ ] Documentation update



<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Added new functions for creating event, entity, and join data sources.
- Introduced enhanced date macro utilities to enable flexible SQL query
substitutions.

- **Refactor**
- Streamlined naming conventions and standardized parameter formatting.
- Consolidated and simplified import structures for improved
consistency.
- Updated method signatures and calls from `select` to `selects` across
various components.
- Removed reliance on `ttypes` for source definitions and standardized
parameter naming conventions.
  - Simplified macro substitution logic in the `StagingQuery` object.

- **Tests**
- Implemented comprehensive tests for date manipulation features to
ensure robust behavior.
- Updated existing tests to reflect changes in method names and query
formatting.
- Adjusted data generation parameters in tests to increase transaction
volumes.

- **Documentation**
- Updated configuration descriptions to clearly illustrate new date
template options and parameter adjustments.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: nikhil-zlai

api/py/ai/chronon/group_by.py

@@ -242,8 +242,8 @@ def normalize(w: Union[common.Window, str]) -> common.Window:
     return agg
 
 
-def Window(length: int, timeUnit: common.TimeUnit) -> common.Window:
-    return common.Window(length, timeUnit)
+def Window(length: int, time_unit: common.TimeUnit) -> common.Window:
+    return common.Window(length, time_unit)
 
 
 def Derivation(name: str, expression: str) -> ttypes.Derivation:

api/py/ai/chronon/query.py

@@ -83,6 +83,23 @@ def Query(
     )
 
 
-def select(*args, **kwargs):
+def selects(*args, **kwargs):
+    """
+    Create a dictionary required for the selects parameter of Query.
+
+    .. code-bloour clients:: python
+        selects(
+            "event_id",
+            user_id="user_id",
+        )
+
+    creates the following dictionary:
+
+    .. code-bloour clients:: python
+        {
+            "event_id": "event_id",
+            "user_id": "user_id"
+        }
+    """
     args = {x: x for x in args}
     return {**args, **kwargs}

api/py/ai/chronon/source.py

@@ -0,0 +1,88 @@
+"""
+Wrappers to directly create Source objects.
+"""
+
+import ai.chronon.api.ttypes as ttypes
+
+
+def EventSource(
+    table: str,
+    query: ttypes.Query,
+    topic: str = None,
+    is_cumulative: bool = None,
+) -> ttypes.Source:
+    """
+    Event Sources represent data that gets generated over-time.
+    Typically, but not necessarily, logged to message buses like kafka, kinesis or google pub/sub.
+    fct tables are also event source worthy.
+
+    Attributes:
+
+     - table: Table currently needs to be a 'ds' (date string - yyyy-MM-dd) partitioned hive table.
+              Table names can contain subpartition specs, example db.table/system=mobile/currency=USD
+     - topic: Topic is a kafka table. The table contains all the events historically came through this topic.
+     - query: The logic used to scan both the table and the topic. Contains row level transformations
+              and filtering expressed as Spark SQL statements.
+     - isCumulative: If each new hive partition contains not just the current day's events but the entire set
+                     of events since the begininng. The key property is that the events are not mutated
+                     across partitions.
+
+    """
+    return ttypes.Source(
+        events=ttypes.EventSource(
+            table=table, topic=topic, query=query, isCumulative=is_cumulative
+        )
+    )
+
+
+def EntitySource(
+    snapshot_table: str,
+    query: ttypes.Query,
+    mutation_table: str = None,
+    mutation_topic: str = None,
+) -> ttypes.Source:
+    """
+    Entity Sources represent data that gets mutated over-time - at row-level. This is a group of three data elements.
+    snapshotTable, mutationTable and mutationTopic. mutationTable and mutationTopic are only necessary if we are trying
+    to create realtime or point-in-time aggregations over these sources. Entity sources usually map 1:1 with a database
+    tables in your OLTP store that typically serves live application traffic. When mutation data is absent they map 1:1
+    to `dim` tables in star schema.
+
+    Attributes:
+     - snapshotTable: Snapshot table currently needs to be a 'ds' (date string - yyyy-MM-dd) partitioned hive table.
+     - mutationTable: Topic is a kafka table. The table contains
+                      all the events that historically came through this topic.
+                      We need all the fields present in the snapshot table, PLUS two additional fields,
+                      `mutation_time` - milliseconds since epoch of type Long that represents the time of the mutation
+                      `is_before` - a boolean flag that represents whether
+                                    this row contains values before or after the mutation.
+     - mutationTopic: The logic used to scan both the table and the topic. Contains row level transformations
+                      and filtering expressed as Spark SQL statements.
+     - query: If each new hive partition contains not just the current day's events but the entire set
+              of events since the begininng. The key property is that the events are not mutated across partitions.
+    """
+    return ttypes.Source(
+        entities=ttypes.EntitySource(
+            snapshotTable=snapshot_table,
+            mutationTable=mutation_table,
+            mutationTopic=mutation_topic,
+            query=query,
+        )
+    )
+
+
+def JoinSource(join: ttypes.Join, query: ttypes.Query) -> ttypes.Source:
+    """
+    The output of a join can be used as a source for `GroupBy`.
+    Useful for expressing complex computation in chronon.
+
+    Offline this simply means that we will compute the necessary date ranges of the join
+    before we start computing the `GroupBy`.
+
+    Online we will:
+    1. enrich the stream/topic of `join.left` with all the columns defined by the join
+    2. apply the selects & wheres defined in the `query`
+    3. perform aggregations defined in the *downstream* `GroupBy`
+    4. write the result to the kv store.
+    """
+    return ttypes.Source(joinSource=ttypes.JoinSource(join=join, query=query))

api/py/ai/chronon/types.py

@@ -0,0 +1,48 @@
+"""
+importing ai.chronon.types will bring in all the api's needed to create any chronon object
+"""
+
+import ai.chronon.group_by as group_by
+import ai.chronon.query as query
+import ai.chronon.join as join
+import ai.chronon.source as source
+import ai.chronon.api.ttypes as ttypes
+
+
+# source related concepts
+Query = query.Query
+selects = query.selects
+
+
+EventSource = source.EventSource
+EntitySource = source.EntitySource
+JoinSource = source.JoinSource
+
+# Aggregation / GroupBy related concepts
+GroupBy = group_by.GroupBy
+Aggregation = group_by.Aggregation
+Operation = group_by.Operation
+Window = group_by.Window
+TimeUnit = group_by.TimeUnit
+DefaultAggregation = group_by.DefaultAggregation
+
+Accuracy = ttypes.Accuracy
+TEMPORAL = ttypes.Accuracy.TEMPORAL
+SNAPSHOT = ttypes.Accuracy.SNAPSHOT
+
+Derivation = group_by.Derivation
+
+# join related concepts
+Join = join.Join
+JoinPart = join.JoinPart
+BootstrapPart = join.BootstrapPart
+LabelParts = join.LabelParts
+ContextualSource = join.ContextualSource
+ExternalPart = join.ExternalPart
+ExternalSource = join.ExternalSource
+DataType = join.DataType
+
+
+# Staging Query related concepts
+StagingQuery = ttypes.StagingQuery
+MetaData = ttypes.MetaData

api/py/test/sample/group_bys/kaggle/clicks.py

@@ -13,7 +13,7 @@
 #     limitations under the License.
 
 from ai.chronon.api.ttypes import Source, EventSource
-from ai.chronon.query import Query, select
+from ai.chronon.query import Query, selects
 from ai.chronon.group_by import (
     GroupBy,
     Aggregation,
@@ -46,7 +46,7 @@
             base_table
         ),  # Here we use the staging query output table because it has the necessary fields, but for a true streaming source we would likely use a log table
         topic="some_topic",  # You would set your streaming source topic here
-        query=Query(selects=select("ad_id", "cliour clientsed"), time_column="ts"),
+        query=Query(selects=selects("ad_id", "cliour clientsed"), time_column="ts"),
     )
 )
 

api/py/test/sample/group_bys/quickstart/purchases.py

@@ -13,7 +13,7 @@
 #     limitations under the License.
 
 from ai.chronon.api.ttypes import Source, EventSource
-from ai.chronon.query import Query, select
+from ai.chronon.query import Query, selects
 from ai.chronon.group_by import GroupBy, Aggregation, Operation
 
 """
@@ -26,10 +26,8 @@
         table="data.purchases",  # This points to the log table in the warehouse with historical purchase events, updated in batch daily
         topic=None,  # See the 'returns' GroupBy for an example that has a streaming source configured. In this case, this would be the streaming source topic that can be listened to for realtime events
         query=Query(
-            selects=select(
-                "user_id",
-                "purchase_price",
-                buour clientset_rand="'1'"
+            selects=selects(
+                "user_id", "purchase_price", buour clientset_rand="'1'"
             ),  # Select the fields we care about
             time_column="ts",
         ),  # The event time

api/py/test/sample/group_bys/quickstart/returns.py

@@ -13,7 +13,7 @@
 #     limitations under the License.
 
 from ai.chronon.api.ttypes import Source, EventSource
-from ai.chronon.query import Query, select
+from ai.chronon.query import Query, selects
 from ai.chronon.group_by import (
     GroupBy,
     Aggregation,
@@ -29,7 +29,7 @@
         table="data.returns",  # This points to the log table with historical return events
         topic="events.returns/fields=ts,return_id,user_id,product_id,refund_amt/host=kafka/port=9092",
         query=Query(
-            selects=select("user_id", "refund_amt"),  # Select the fields we care about
+            selects=selects("user_id", "refund_amt"),  # Select the fields we care about
             time_column="ts",
         ),  # The event time
     )

api/py/test/sample/group_bys/quickstart/schema.py

@@ -1,15 +1,15 @@
 from ai.chronon.group_by import GroupBy, Aggregation, Operation
 from ai.chronon.api.ttypes import Source, EventSource
-from ai.chronon.query import Query, select
+from ai.chronon.query import Query, selects
 
 
 logging_schema_source = Source(
     events=EventSource(
         table="default.chronon_log_table",
         query=Query(
-            selects=select(
+            selects=selects(
                 schema_hash="decode(unbase64(key_base64), 'utf-8')",
-                schema_value="decode(unbase64(value_base64), 'utf-8')"
+                schema_value="decode(unbase64(value_base64), 'utf-8')",
             ),
             wheres=["name='SCHEMA_PUBLISH_EVENT'"],
             time_column="ts_millis",
@@ -20,12 +20,7 @@
 v1 = GroupBy(
     keys=["schema_hash"],
     sources=logging_schema_source,
-    aggregations=[
-        Aggregation(
-            input_column="schema_value",
-            operation=Operation.LAST
-        )
-    ],
+    aggregations=[Aggregation(input_column="schema_value", operation=Operation.LAST)],
     online=False,
-    baour clientsfill_start_date="2023-04-09"
+    baour clientsfill_start_date="2023-04-09",
 )

api/py/test/sample/group_bys/quickstart/users.py

@@ -1,4 +1,3 @@
-
 #     Copyright (C) 2023 The Chronon Authors.
 #
 #     Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,7 +13,7 @@
 #     limitations under the License.
 
 from ai.chronon.api.ttypes import Source, EntitySource
-from ai.chronon.query import Query, select
+from ai.chronon.query import Query, selects
 from ai.chronon.group_by import (
     GroupBy,
 )
@@ -26,15 +25,18 @@
 
 source = Source(
     entities=EntitySource(
-        snapshotTable="data.users", # This points to a table that contains daily snapshots of the entire product catalog
+        snapshotTable="data.users",  # This points to a table that contains daily snapshots of the entire product catalog
         query=Query(
-            selects=select("user_id","account_created_ds","email_verified"), # Select the fields we care about
-        )
-    ))
+            selects=selects(
+                "user_id", "account_created_ds", "email_verified"
+            ),  # Select the fields we care about
+        ),
+    )
+)
 
 v1 = GroupBy(
     sources=[source],
-    keys=["user_id"], # Primary key is the same as the primary key for the source table
-    aggregations=None, # In this case, there are no aggregations or windows to define
+    keys=["user_id"],  # Primary key is the same as the primary key for the source table
+    aggregations=None,  # In this case, there are no aggregations or windows to define
     online=True,
-) 
+)

api/py/test/sample/group_bys/risk/merchant_data.py

@@ -1,8 +1,6 @@
 from ai.chronon.api.ttypes import Source, EntitySource
-from ai.chronon.query import Query, select
-from ai.chronon.group_by import (
-    GroupBy
-)
+from ai.chronon.query import Query, selects
+from ai.chronon.group_by import GroupBy
 
 """
 This GroupBy aggregates metrics about a user's previous purchases in various windows.
@@ -11,15 +9,21 @@
 # This source is raw purchase events. Every time a user makes a purchase, it will be one entry in this source.
 source_merchants = Source(
     entities=EntitySource(
-        snapshotTable="data.merchants", # This points to the log table in the warehouse with historical purchase events, updated in batch daily
+        snapshotTable="data.merchants",  # This points to the log table in the warehouse with historical purchase events, updated in batch daily
         query=Query(
-            selects=select("merchant_id","account_age", "zipcode", "is_big_merchant", "country", "account_type", "preferred_language"), # Select the fields we care about
-        )
+            selects=selects(
+                "merchant_id",
+                "account_age",
+                "zipcode",
+                "is_big_merchant",
+                "country",
+                "account_type",
+                "preferred_language",
+            ),  # Select the fields we care about
+        ),
     )
 )
 
 merchant_group_by = GroupBy(
-    sources=[source_merchants],
-    keys=["merchant_id"],
-    aggregations=None
-)
+    sources=[source_merchants], keys=["merchant_id"], aggregations=None
+)

api/py/test/sample/group_bys/risk/transaction_events.py

@@ -1,5 +1,5 @@
 from ai.chronon.api.ttypes import Source, EventSource
-from ai.chronon.query import Query, select
+from ai.chronon.query import Query, selects
 from ai.chronon.group_by import GroupBy, Aggregation, Operation
 
 """
@@ -13,7 +13,7 @@ def create_transaction_source(key_field):
             table="data.txn_events",  # Points to the historical purchase events table
             topic=None,
             query=Query(
-                selects=select(key_field, "transaction_amount", "transaction_type"),
+                selects=selects(key_field, "transaction_amount", "transaction_type"),
                 time_column="transaction_time",
             ),
         )