fix: Added concurrency groups on workflows

ntkathole · ntkathole · commit 0b58224a0a64 · 2026-03-03T15:31:01.000+05:30
Signed-off-by: ntkathole &lt;nikhilkathole2683@gmail.com&gt;
diff --git a/.github/workflows/dbt-integration-tests.yml b/.github/workflows/dbt-integration-tests.yml
@@ -13,6 +13,10 @@ on:
       - 'sdk/python/tests/unit/dbt/**'
       - '.github/workflows/dbt-integration-tests.yml'
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   dbt-integration-test:
     if:
diff --git a/.github/workflows/lint_pr.yml b/.github/workflows/lint_pr.yml
@@ -7,6 +7,10 @@ on:
       - edited
       - synchronize
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   validate-title:
     if:
diff --git a/.github/workflows/linter.yml b/.github/workflows/linter.yml
@@ -2,6 +2,10 @@ name: linter
 
 on: [push, pull_request]
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   lint-python:
     runs-on: [ubuntu-latest]
diff --git a/.github/workflows/operator-e2e-integration-tests.yml b/.github/workflows/operator-e2e-integration-tests.yml
@@ -13,6 +13,10 @@ on:
     paths:
       - 'infra/**'
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   operator-e2e-tests:
     timeout-minutes: 40
diff --git a/.github/workflows/operator_pr.yml b/.github/workflows/operator_pr.yml
@@ -1,6 +1,11 @@
 name: operator-pr
 
 on: [pull_request]
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   operator-test:
     runs-on: ubuntu-latest
diff --git a/.github/workflows/pr_duckdb_integration_tests.yml b/.github/workflows/pr_duckdb_integration_tests.yml
@@ -7,6 +7,10 @@ on:
       - synchronize
       - labeled
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   integration-test-duckdb-offline:
     if:
diff --git a/.github/workflows/pr_integration_tests.yml b/.github/workflows/pr_integration_tests.yml
@@ -7,10 +7,10 @@ on:
       - synchronize
       - labeled
 
-# concurrency is currently broken, see details https://github.com/actions/runner/issues/1532
-#concurrency:
-#  group: pr-integration-tests-${{ github.event.pull_request.number }}
-#  cancel-in-progress: true
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 permissions:
   actions: write
   pull-requests: read
diff --git a/.github/workflows/pr_local_integration_tests.yml b/.github/workflows/pr_local_integration_tests.yml
@@ -8,6 +8,10 @@ on:
       - synchronize
       - labeled
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   integration-test-python-local:
     if:
diff --git a/.github/workflows/pr_ray_integration_tests.yml b/.github/workflows/pr_ray_integration_tests.yml
@@ -7,6 +7,10 @@ on:
       - synchronize
       - labeled
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   integration-test-ray:
     if:
diff --git a/.github/workflows/pr_registration_integration_tests.yml b/.github/workflows/pr_registration_integration_tests.yml
@@ -7,6 +7,10 @@ on:
       - synchronize
       - labeled
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 permissions:
   actions: write
   pull-requests: read
@@ -39,6 +43,16 @@ jobs:
       (github.event.action != 'labeled' && (contains(github.event.pull_request.labels.*.name, 'ok-to-test') || contains(github.event.pull_request.labels.*.name, 'approved') || contains(github.event.pull_request.labels.*.name, 'lgtm')))) &&
       github.repository == 'feast-dev/feast'
     runs-on: ubuntu-latest
+    services:
+      redis:
+        image: redis
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     steps:
       - uses: actions/checkout@v4
         with:
diff --git a/.github/workflows/pr_remote_rbac_integration_tests.yml b/.github/workflows/pr_remote_rbac_integration_tests.yml
@@ -8,6 +8,10 @@ on:
       - synchronize
       - labeled
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   remote-rbac-integration-tests-python:
     if:
diff --git a/.github/workflows/registry-rest-api-tests.yml b/.github/workflows/registry-rest-api-tests.yml
@@ -11,6 +11,10 @@ on:
       - synchronize
       - labeled
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   registry-rest-api-tests:
     timeout-minutes: 30
diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml
@@ -8,6 +8,10 @@ on:
   schedule:
     - cron: "0 6 * * 1"
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   codeql:
     name: CodeQL Analysis
diff --git a/.github/workflows/smoke_tests.yml b/.github/workflows/smoke_tests.yml
@@ -3,6 +3,10 @@ name: smoke-tests
 on:
   pull_request:
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   smoke-test-python:
     runs-on: ${{ matrix.os }}
diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml
@@ -6,6 +6,10 @@ on:
     branches:
       - master
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   unit-test-python:
     runs-on: ${{ matrix.os }}
diff --git a/Makefile b/Makefile
@@ -184,7 +184,7 @@ test-python-smoke: ## Quick smoke test for development
 test-python-integration: ## Run Python integration tests (CI)
 	uv run python -m pytest --tb=short -v -n 8 --integration --color=yes --durations=10 --timeout=1200 --timeout_method=thread --dist loadgroup \
 		-k "(not snowflake or not test_historical_features_main)" \
-		-m "not rbac_remote_integration_test" \
+		-m "not rbac_remote_integration_test and not ray_offline_stores_only" \
 		--ignore=sdk/python/tests/integration/registration \
 		--ignore=sdk/python/tests/integration/compute_engines/ray_compute \
 		--log-cli-level=INFO -s \
diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py
@@ -833,12 +833,17 @@ def arrow_schema_to_bq_schema(arrow_schema: pyarrow.Schema) -> List[SchemaField]
     bq_schema = []
 
     for field in arrow_schema:
-        if pyarrow.types.is_list(field.type):
+        if pyarrow.types.is_struct(field.type) or pyarrow.types.is_map(field.type):
+            detected_mode = "NULLABLE"
+            detected_type = "STRING"
+        elif pyarrow.types.is_list(field.type):
             detected_mode = "REPEATED"
-            detected_type = _ARROW_SCALAR_IDS_TO_BQ[field.type.value_type.id]
+            detected_type = _ARROW_SCALAR_IDS_TO_BQ.get(
+                field.type.value_type.id, "STRING"
+            )
         else:
             detected_mode = "NULLABLE"
-            detected_type = _ARROW_SCALAR_IDS_TO_BQ[field.type.id]
+            detected_type = _ARROW_SCALAR_IDS_TO_BQ.get(field.type.id, "STRING")
 
         bq_schema.append(
             SchemaField(name=field.name, field_type=detected_type, mode=detected_mode)
diff --git a/sdk/python/feast/infra/online_stores/milvus_online_store/milvus.py b/sdk/python/feast/infra/online_stores/milvus_online_store/milvus.py
@@ -87,6 +87,10 @@
             FEAST_PRIMITIVE_TO_MILVUS_TYPE_MAPPING[feast_type] = DataType.VARCHAR
         elif base_value_type == ValueType.BOOL:
             FEAST_PRIMITIVE_TO_MILVUS_TYPE_MAPPING[feast_type] = DataType.BINARY_VECTOR
+    elif isinstance(feast_type, ComplexFeastType):
+        milvus_type = PROTO_TO_MILVUS_TYPE_MAPPING.get(value_type)
+        if milvus_type:
+            FEAST_PRIMITIVE_TO_MILVUS_TYPE_MAPPING[feast_type] = milvus_type
 
 
 class MilvusOnlineStoreConfig(FeastConfigBaseModel, VectorStoreConfig):
@@ -179,6 +183,8 @@ def _get_or_create_collection(
             fields_to_add = [f for f in table.schema if f.name not in fields_to_exclude]
             for field in fields_to_add:
                 dtype = FEAST_PRIMITIVE_TO_MILVUS_TYPE_MAPPING.get(field.dtype)
+                if dtype is None and isinstance(field.dtype, ComplexFeastType):
+                    dtype = DataType.VARCHAR
                 if dtype:
                     if dtype == DataType.FLOAT_VECTOR:
                         fields.append(
diff --git a/sdk/python/feast/infra/online_stores/remote.py b/sdk/python/feast/infra/online_stores/remote.py
@@ -94,6 +94,15 @@ def _proto_value_to_transport_value(proto_value: ValueProto) -> Any:
         if val_attr == "json_list_val":
             return list(getattr(proto_value, val_attr).val)
 
+        # Map/Struct types are converted to Python dicts by
+        # feast_value_type_to_python_type.  Serialise them to JSON strings
+        # so the server-side DataFrame gets VARCHAR columns instead of
+        # PyArrow struct columns that can crash with complex nested types.
+        if val_attr in ("map_val", "struct_val"):
+            return json.dumps(feast_value_type_to_python_type(proto_value))
+        if val_attr in ("map_list_val", "struct_list_val"):
+            return [json.dumps(v) for v in feast_value_type_to_python_type(proto_value)]
+
         return feast_value_type_to_python_type(proto_value)
 
     def online_write_batch(
diff --git a/sdk/python/feast/infra/utils/aws_utils.py b/sdk/python/feast/infra/utils/aws_utils.py
@@ -243,7 +243,6 @@ def upload_df_to_redshift(
     Raises:
         RedshiftTableNameTooLong: The specified table name is too long.
     """
-
     # Drop the index so that we dont have unnecessary columns
     df.reset_index(drop=True, inplace=True)
 
diff --git a/sdk/python/tests/conftest.py b/sdk/python/tests/conftest.py
@@ -435,6 +435,9 @@ def fake_ingest_data():
         "conv_rate": [0.5],
         "acc_rate": [0.6],
         "avg_daily_trips": [4],
+        "driver_metadata": [None],
+        "driver_config": [None],
+        "driver_profile": [None],
         "event_timestamp": [pd.Timestamp(_utc_now()).round("ms")],
         "created": [pd.Timestamp(_utc_now()).round("ms")],
     }
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_source_creator.py b/sdk/python/tests/integration/feature_repos/universal/data_source_creator.py
@@ -1,3 +1,4 @@
+import json
 from abc import ABC, abstractmethod
 from typing import Dict, Optional
 
@@ -14,6 +15,29 @@ class DataSourceCreator(ABC):
     def __init__(self, project_name: str, *args, **kwargs):
         self.project_name = project_name
 
+    @staticmethod
+    def serialize_complex_columns(df: pd.DataFrame) -> pd.DataFrame:
+        """Serialize dict columns (Map/Struct types) to JSON strings.
+
+        Backends like Snowflake, BigQuery, and Redshift cannot natively
+        ingest Python dicts via their bulk-load paths (VARIANT, STRUCT,
+        super types cause issues).  Converting them to JSON strings lets
+        the data be stored as VARCHAR/STRING instead.
+
+        List columns with primitive values (int, float, str, bool) are
+        left untouched since backends handle those as native ARRAY types.
+        """
+        df = df.copy()
+        for col in df.columns:
+            if df[col].dropna().empty:
+                continue
+            sample = df[col].dropna().iloc[0]
+            if isinstance(sample, dict):
+                df[col] = df[col].apply(
+                    lambda v: json.dumps(v) if v is not None else None
+                )
+        return df
+
     @abstractmethod
     def create_data_source(
         self,
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/bigquery.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/bigquery.py
@@ -80,6 +80,7 @@ def create_data_source(
         # `BigQueryOfflineStore.offline_write_batch`, but since we're bypassing that API here, we should follow the same
         # rule. The schema of this initial dataframe determines the schema in the newly created BigQuery table.
         df = make_df_tzaware(df)
+        df = self.serialize_complex_columns(df)
         job = self.client.load_table_from_dataframe(df, destination_name)
         job.result()
 
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/redshift.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/redshift.py
@@ -55,6 +55,7 @@ def create_data_source(
     ) -> DataSource:
         destination_name = self.get_prefixed_table_name(destination_name)
 
+        df = self.serialize_complex_columns(df)
         aws_utils.upload_df_to_redshift(
             self.client,
             self.offline_store_config.cluster_id,
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/snowflake.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/snowflake.py
@@ -53,6 +53,7 @@ def create_data_source(
     ) -> DataSource:
         destination_name = self.get_prefixed_table_name(destination_name)
 
+        df = self.serialize_complex_columns(df)
         with GetSnowflakeConnection(self.offline_store_config) as conn:
             write_pandas(conn, df, destination_name, auto_create_table=True)
 
diff --git a/sdk/python/tests/integration/offline_store/test_offline_write.py b/sdk/python/tests/integration/offline_store/test_offline_write.py
@@ -38,16 +38,16 @@ def test_reorder_columns(environment, universal_data_sources):
             "acc_rate": [random.random(), random.random()],
             "driver_id": [1001, 1001],
             "driver_metadata": [
-                {"vehicle_type": "sedan", "rating": "4.5"},
-                {"vehicle_type": "suv", "rating": "3.8"},
+                json.dumps({"vehicle_type": "sedan", "rating": "4.5"}),
+                json.dumps({"vehicle_type": "suv", "rating": "3.8"}),
             ],
             "driver_config": [
                 json.dumps({"max_distance_km": 100, "preferred_zones": ["north"]}),
                 json.dumps({"max_distance_km": 50, "preferred_zones": ["south"]}),
             ],
             "driver_profile": [
-                {"name": "driver_1001", "age": "30"},
-                {"name": "driver_1001", "age": "30"},
+                json.dumps({"name": "driver_1001", "age": "30"}),
+                json.dumps({"name": "driver_1001", "age": "30"}),
             ],
         },
     )
@@ -158,16 +158,16 @@ def test_writing_consecutively_to_offline_store(environment, universal_data_sour
             "avg_daily_trips": [random.randint(0, 10), random.randint(0, 10)],
             "created": [ts, ts],
             "driver_metadata": [
-                {"vehicle_type": "sedan", "rating": "4.5"},
-                {"vehicle_type": "suv", "rating": "3.8"},
+                json.dumps({"vehicle_type": "sedan", "rating": "4.5"}),
+                json.dumps({"vehicle_type": "suv", "rating": "3.8"}),
             ],
             "driver_config": [
                 json.dumps({"max_distance_km": 100, "preferred_zones": ["north"]}),
                 json.dumps({"max_distance_km": 50, "preferred_zones": ["south"]}),
             ],
             "driver_profile": [
-                {"name": "driver_1001", "age": "30"},
-                {"name": "driver_1001", "age": "35"},
+                json.dumps({"name": "driver_1001", "age": "30"}),
+                json.dumps({"name": "driver_1001", "age": "35"}),
             ],
         },
     )
@@ -214,16 +214,16 @@ def test_writing_consecutively_to_offline_store(environment, universal_data_sour
             "avg_daily_trips": [random.randint(0, 10), random.randint(0, 10)],
             "created": [ts, ts],
             "driver_metadata": [
-                {"vehicle_type": "truck", "rating": "4.0"},
-                {"vehicle_type": "sedan", "rating": "4.2"},
+                json.dumps({"vehicle_type": "truck", "rating": "4.0"}),
+                json.dumps({"vehicle_type": "sedan", "rating": "4.2"}),
             ],
             "driver_config": [
                 json.dumps({"max_distance_km": 150, "preferred_zones": ["east"]}),
                 json.dumps({"max_distance_km": 200, "preferred_zones": ["west"]}),
             ],
             "driver_profile": [
-                {"name": "driver_1001", "age": "31"},
-                {"name": "driver_1001", "age": "36"},
+                json.dumps({"name": "driver_1001", "age": "31"}),
+                json.dumps({"name": "driver_1001", "age": "36"}),
             ],
         },
     )
