Add label join flag to custom json for airflow (#775)

## Summary

Adding a flag so that airflow integration knows whether to schedule a
join or not

## Checklist
- [ ] 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**
- Enhanced join metadata to include a flag indicating the presence of
label parts.
- **Tests**
- Updated sample join test to include label part information in join
instantiation.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: ezvz <[email protected]>

Author: varant-zlai

api/python/ai/chronon/airflow_helpers.py

@@ -149,6 +149,9 @@ def _set_join_deps(join):
     # Update the metadata customJson with dependencies
     _dedupe_and_set_airflow_deps_json(join, deps)
 
+    # Set the t/f flag for label_join
+    _set_label_join_flag(join)
+
 
 def _set_group_by_deps(group_by):
     if not group_by.sources:
@@ -170,6 +173,16 @@ def _set_group_by_deps(group_by):
     _dedupe_and_set_airflow_deps_json(group_by, deps)
 
 
+def _set_label_join_flag(join):
+    existing_json = join.metaData.customJson or "{}"
+    json_map = json.loads(existing_json)
+    label_join_flag = False
+    if join.labelParts:
+        label_join_flag = True
+    json_map["label_join"] = label_join_flag
+    join.metaData.customJson = json.dumps(json_map)
+
+
 def _dedupe_and_set_airflow_deps_json(obj, deps):
     sorted_items = [tuple(sorted(d.items())) for d in deps]
     # Use OrderedDict for re-producible ordering of dependencies

api/python/test/sample/joins/sample_team/sample_join.py

@@ -20,7 +20,7 @@
     JoinPart,
 )
 from ai.chronon.repo.constants import RunMode
-from ai.chronon.types import EnvironmentVariables
+from ai.chronon.types import EnvironmentVariables, LabelParts
 
 v1 = Join(
     left=test_sources.staging_entities,
@@ -33,6 +33,7 @@
         }
     ),
     online=True,
+    label_part=LabelParts([JoinPart(group_by=sample_group_by.v1)], 1, 1)
 )
 
 never = Join(