feast-dev
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/pixi.lock b/pixi.lock
diff --git a/pyproject.toml b/pyproject.toml
@@ -293,7 +293,7 @@ test-compute = { cmd = "python -m pytest -v --integration sdk/python/tests/integ
 test = { depends-on = ["test-offline", "test-compute"] }
 
 [tool.pixi.feature.registration-tests.pypi-dependencies]
-feast = { path = ".", editable = true, extras = ["aws", "gcp", "grpcio", "postgres", "mysql", "snowflake", "spark", "test"] }
+feast = { path = ".", editable = true, extras = ["aws", "gcp", "grpcio", "postgres", "mysql", "redis", "snowflake", "spark", "test"] }
 grpcio-testing = ">=1.56.2,<=1.62.3"
 
 [tool.pixi.feature.registration-tests.tasks]

@@ -416,7 +416,7 @@ def offline_write_batch(
             )
 
         if table.schema != pa_schema:
-            table = table.cast(pa_schema)
+            table = offline_utils.cast_arrow_table_to_schema(table, pa_schema)
         project_id = (
             config.offline_store.billing_project_id or config.offline_store.project_id
         )
@@ -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)

@@ -260,6 +260,37 @@ def get_pyarrow_schema_from_batch_source(
     return pa.schema(pa_schema), column_names
 
 
+def cast_arrow_table_to_schema(table: pa.Table, pa_schema: pa.Schema) -> pa.Table:
+    """Cast a PyArrow table to match the target schema, handling struct/map → string.
+
+    PyArrow cannot natively cast struct or map columns to string.  When a
+    SQL-based offline store (BigQuery, Snowflake, Redshift) stores complex
+    Feast types (Map, Struct) as VARCHAR/STRING, the target schema will have
+    string fields while the input table may have struct/map fields (e.g. when
+    the caller provides Python dicts).  This function serialises those columns
+    to JSON strings so the subsequent cast succeeds.
+    """
+    import json as _json
+
+    for i, field in enumerate(table.schema):
+        target_type = pa_schema.field(field.name).type
+        is_complex_source = pa.types.is_struct(field.type) or pa.types.is_map(
+            field.type
+        )
+        is_string_target = pa.types.is_string(target_type) or pa.types.is_large_string(
+            target_type
+        )
+        if is_complex_source and is_string_target:
+            col = table.column(i)
+            json_arr = pa.array(
+                [_json.dumps(v.as_py()) if v.is_valid else None for v in col],
+                type=target_type,
+            )
+            table = table.set_column(i, field.name, json_arr)
+
+    return table.cast(pa_schema)
+
+
 def enclose_in_backticks(value):
     # Check if the input is a list
     if isinstance(value, list):

@@ -353,7 +353,7 @@ def offline_write_batch(
             )
 
         if table.schema != pa_schema:
-            table = table.cast(pa_schema)
+            table = offline_utils.cast_arrow_table_to_schema(table, pa_schema)
 
         redshift_options = feature_view.batch_source.redshift_options
         redshift_client = aws_utils.get_redshift_data_client(

@@ -409,7 +409,7 @@ def offline_write_batch(
             )
 
         if table.schema != pa_schema:
-            table = table.cast(pa_schema)
+            table = offline_utils.cast_arrow_table_to_schema(table, pa_schema)
 
         with GetSnowflakeConnection(config.offline_store) as conn:
             snowflake_conn = conn