fix: Added concurrency groups on workflows

ntkathole · ntkathole · commit f96af4971573 · 2026-03-03T14:38: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
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/utils/aws_utils.py b/sdk/python/feast/infra/utils/aws_utils.py
@@ -247,6 +247,15 @@ def upload_df_to_redshift(
     # Drop the index so that we dont have unnecessary columns
     df.reset_index(drop=True, inplace=True)
 
+    # Serialize complex columns (dicts/lists) to JSON strings so they map to
+    # varchar instead of Redshift's super type, which COPY FROM PARQUET cannot handle.
+    import json as _json
+
+    for col in df.columns:
+        non_null = df[col].dropna()
+        if not non_null.empty and isinstance(non_null.iloc[0], (dict, list)):
+            df[col] = df[col].apply(lambda v: _json.dumps(v) if v is not None else None)
+
     # Convert Pandas DataFrame into PyArrow table and compile the Redshift table schema.
     # Note, if the underlying data has missing values,
     # pandas will convert those values to np.nan if the dtypes are numerical (floats, ints, etc.) or boolean.
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,26 @@ 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/list columns to JSON strings.
+
+        Backends like Snowflake, BigQuery, and Redshift cannot natively
+        ingest Python dicts or lists 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.
+        """
+        df = df.copy()
+        for col in df.columns:
+            if df[col].dropna().empty:
+                continue
+            sample = df[col].dropna().iloc[0]
+            if isinstance(sample, (dict, list)):
+                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/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)
 
