Skip to content

Commit f2ca4bd

Browse files
claudeclaude
committed
feat: Add s3_url_style configuration for DuckDB offline store
This adds support for configuring S3 URL style (path vs vhost) for DuckDB offline store when working with S3-compatible storage systems like MinIO or LocalStack that require path-style URLs. Changes: - Add s3_url_style field to FileOptions proto message - Add s3_url_style to FileDestination proto message - Add staging_location_s3_url_style to DuckDBOfflineStoreConfig - Add s3_url_style parameter to FileSource, FileOptions, SavedDatasetFileStorage, and FileLoggingDestination - Create _build_s3_storage_options helper function for consistent storage options handling - Update IbisRetrievalJob to support s3_url_style for remote storage - Update test data source creators to use path-style URLs Fixes feast-dev#5597 Signed-off-by: Claude <[email protected]>
1 parent 59dbb33 commit f2ca4bd

10 files changed

Lines changed: 391 additions & 187 deletions

File tree

‎protos/feast/core/DataSource.proto‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,10 @@ message DataSource {
111111

112112
// override AWS S3 storage endpoint with custom S3 endpoint
113113
string s3_endpoint_override = 3;
114+
115+
// S3 URL style for path-based access (required for MinIO, LocalStack, etc.)
116+
// Valid values: "path" or "vhost" (virtual-hosted style, default)
117+
string s3_url_style = 4;
114118
}
115119

116120
// Defines options for DataSource that sources features from a BigQuery Query

‎protos/feast/core/FeatureService.proto‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,10 @@ message LoggingConfig {
7070

7171
// column names to use for partitioning
7272
repeated string partition_by = 3;
73+
74+
// S3 URL style for path-based access (required for MinIO, LocalStack, etc.)
75+
// Valid values: "path" or "vhost" (virtual-hosted style, default)
76+
string s3_url_style = 4;
7377
}
7478

7579
message BigQueryDestination {

‎sdk/python/feast/infra/offline_stores/duckdb.py‎

Lines changed: 28 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -27,15 +27,29 @@
2727
from feast.repo_config import FeastConfigBaseModel, RepoConfig
2828

2929

30+
def _build_s3_storage_options(
31+
endpoint_override: Optional[str] = None,
32+
s3_url_style: Optional[str] = None,
33+
) -> dict:
34+
"""Build storage_options dict for S3 access, only including set values."""
35+
storage_options = {}
36+
if endpoint_override:
37+
storage_options["AWS_ENDPOINT_URL"] = endpoint_override
38+
if s3_url_style:
39+
storage_options["AWS_S3_URL_STYLE"] = s3_url_style
40+
return storage_options
41+
42+
3043
def _read_data_source(data_source: DataSource, repo_path: str) -> Table:
3144
assert isinstance(data_source, FileSource)
3245

3346
if isinstance(data_source.file_format, ParquetFormat):
3447
return ibis.read_parquet(data_source.path)
3548
elif isinstance(data_source.file_format, DeltaFormat):
36-
storage_options = {
37-
"AWS_ENDPOINT_URL": data_source.s3_endpoint_override,
38-
}
49+
storage_options = _build_s3_storage_options(
50+
endpoint_override=data_source.s3_endpoint_override,
51+
s3_url_style=data_source.s3_url_style,
52+
)
3953

4054
return ibis.read_delta(data_source.path, storage_options=storage_options)
4155

@@ -86,9 +100,10 @@ def _write_data_source(
86100
new_table = pyarrow.concat_tables([table, prev_table])
87101
ibis.memtable(new_table).to_parquet(file_options.uri)
88102
elif isinstance(data_source.file_format, DeltaFormat):
89-
storage_options = {
90-
"AWS_ENDPOINT_URL": str(data_source.s3_endpoint_override),
91-
}
103+
storage_options = _build_s3_storage_options(
104+
endpoint_override=data_source.s3_endpoint_override,
105+
s3_url_style=data_source.s3_url_style,
106+
)
92107

93108
if mode == "append":
94109
from deltalake import DeltaTable
@@ -120,6 +135,10 @@ class DuckDBOfflineStoreConfig(FeastConfigBaseModel):
120135

121136
staging_location_endpoint_override: Optional[str] = None
122137

138+
# S3 URL style for path-based access (required for MinIO, LocalStack, etc.)
139+
# Valid values: "path" or "vhost" (virtual-hosted style, default)
140+
staging_location_s3_url_style: Optional[str] = None
141+
123142

124143
class DuckDBOfflineStore(OfflineStore):
125144
@staticmethod
@@ -146,6 +165,7 @@ def pull_latest_from_table_or_query(
146165
data_source_writer=_write_data_source,
147166
staging_location=config.offline_store.staging_location,
148167
staging_location_endpoint_override=config.offline_store.staging_location_endpoint_override,
168+
staging_location_s3_url_style=config.offline_store.staging_location_s3_url_style,
149169
)
150170

151171
@staticmethod
@@ -170,6 +190,7 @@ def get_historical_features(
170190
data_source_writer=_write_data_source,
171191
staging_location=config.offline_store.staging_location,
172192
staging_location_endpoint_override=config.offline_store.staging_location_endpoint_override,
193+
staging_location_s3_url_style=config.offline_store.staging_location_s3_url_style,
173194
)
174195

175196
@staticmethod
@@ -196,6 +217,7 @@ def pull_all_from_table_or_query(
196217
data_source_writer=_write_data_source,
197218
staging_location=config.offline_store.staging_location,
198219
staging_location_endpoint_override=config.offline_store.staging_location_endpoint_override,
220+
staging_location_s3_url_style=config.offline_store.staging_location_s3_url_style,
199221
)
200222

201223
@staticmethod

‎sdk/python/feast/infra/offline_stores/file_source.py‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ def __init__(
4242
created_timestamp_column: Optional[str] = "",
4343
field_mapping: Optional[Dict[str, str]] = None,
4444
s3_endpoint_override: Optional[str] = None,
45+
s3_url_style: Optional[str] = None,
4546
description: Optional[str] = "",
4647
tags: Optional[Dict[str, str]] = None,
4748
owner: Optional[str] = "",
@@ -61,6 +62,8 @@ def __init__(
6162
field_mapping: A dictionary mapping of column names in this data source to feature names in a feature table
6263
or view. Only used for feature columns, not entities or timestamp columns.
6364
s3_endpoint_override (optional): Overrides AWS S3 enpoint with custom S3 storage
65+
s3_url_style (optional): S3 URL style for path-based access. Valid values: "path" or "vhost".
66+
Required for S3-compatible storage like MinIO or LocalStack.
6467
description (optional): A human-readable description.
6568
tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
6669
owner (optional): The owner of the file source, typically the email of the primary
@@ -76,6 +79,7 @@ def __init__(
7679
file_format=file_format,
7780
uri=path,
7881
s3_endpoint_override=s3_endpoint_override,
82+
s3_url_style=s3_url_style,
7983
)
8084

8185
super().__init__(
@@ -102,6 +106,7 @@ def __eq__(self, other):
102106
and self.file_options.file_format == other.file_options.file_format
103107
and self.file_options.s3_endpoint_override
104108
== other.file_options.s3_endpoint_override
109+
and self.file_options.s3_url_style == other.file_options.s3_url_style
105110
)
106111

107112
@property
@@ -119,6 +124,11 @@ def s3_endpoint_override(self) -> Optional[str]:
119124
"""Returns the s3 endpoint override of this file data source."""
120125
return self.file_options.s3_endpoint_override
121126

127+
@property
128+
def s3_url_style(self) -> Optional[str]:
129+
"""Returns the s3 URL style of this file data source."""
130+
return self.file_options.s3_url_style
131+
122132
@staticmethod
123133
def from_proto(data_source: DataSourceProto):
124134
return FileSource(
@@ -129,6 +139,7 @@ def from_proto(data_source: DataSourceProto):
129139
timestamp_field=data_source.timestamp_field,
130140
created_timestamp_column=data_source.created_timestamp_column,
131141
s3_endpoint_override=data_source.file_options.s3_endpoint_override,
142+
s3_url_style=data_source.file_options.s3_url_style or None,
132143
description=data_source.description,
133144
tags=dict(data_source.tags),
134145
owner=data_source.owner,
@@ -201,6 +212,8 @@ def get_table_column_names_and_types(
201212
storage_options = {
202213
"AWS_ENDPOINT_URL": str(self.s3_endpoint_override),
203214
}
215+
if self.s3_url_style:
216+
storage_options["AWS_S3_URL_STYLE"] = self.s3_url_style
204217

205218
schema = (
206219
DeltaTable(self.path, storage_options=storage_options)
@@ -235,18 +248,21 @@ class FileOptions:
235248
Attributes:
236249
uri: File source url, e.g. s3:// or local file.
237250
s3_endpoint_override: Custom s3 endpoint (used only with s3 uri).
251+
s3_url_style: S3 URL style for path-based access ("path" or "vhost").
238252
file_format: File source format, e.g. parquet.
239253
"""
240254

241255
uri: str
242256
file_format: Optional[FileFormat]
243257
s3_endpoint_override: str
258+
s3_url_style: str
244259

245260
def __init__(
246261
self,
247262
uri: str,
248263
file_format: Optional[FileFormat],
249264
s3_endpoint_override: Optional[str],
265+
s3_url_style: Optional[str] = None,
250266
):
251267
"""
252268
Initializes a FileOptions object.
@@ -255,10 +271,12 @@ def __init__(
255271
uri: File source url, e.g. s3:// or local file.
256272
file_format (optional): File source format, e.g. parquet.
257273
s3_endpoint_override (optional): Custom s3 endpoint (used only with s3 uri).
274+
s3_url_style (optional): S3 URL style for path-based access ("path" or "vhost").
258275
"""
259276
self.uri = uri
260277
self.file_format = file_format
261278
self.s3_endpoint_override = s3_endpoint_override or ""
279+
self.s3_url_style = s3_url_style or ""
262280

263281
@classmethod
264282
def from_proto(cls, file_options_proto: DataSourceProto.FileOptions):
@@ -275,6 +293,7 @@ def from_proto(cls, file_options_proto: DataSourceProto.FileOptions):
275293
file_format=FileFormat.from_proto(file_options_proto.file_format),
276294
uri=file_options_proto.uri,
277295
s3_endpoint_override=file_options_proto.s3_endpoint_override,
296+
s3_url_style=file_options_proto.s3_url_style or None,
278297
)
279298
return file_options
280299

@@ -291,6 +310,7 @@ def to_proto(self) -> DataSourceProto.FileOptions:
291310
),
292311
uri=self.uri,
293312
s3_endpoint_override=self.s3_endpoint_override,
313+
s3_url_style=self.s3_url_style,
294314
)
295315

296316
return file_options_proto
@@ -306,10 +326,12 @@ def __init__(
306326
path: str,
307327
file_format: FileFormat = ParquetFormat(),
308328
s3_endpoint_override: Optional[str] = None,
329+
s3_url_style: Optional[str] = None,
309330
):
310331
self.file_options = FileOptions(
311332
file_format=file_format,
312333
s3_endpoint_override=s3_endpoint_override,
334+
s3_url_style=s3_url_style,
313335
uri=path,
314336
)
315337

@@ -320,6 +342,7 @@ def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:
320342
path=file_options.uri,
321343
file_format=file_options.file_format,
322344
s3_endpoint_override=file_options.s3_endpoint_override,
345+
s3_url_style=file_options.s3_url_style or None,
323346
)
324347

325348
def to_proto(self) -> SavedDatasetStorageProto:
@@ -330,6 +353,7 @@ def to_data_source(self) -> DataSource:
330353
path=self.file_options.uri,
331354
file_format=self.file_options.file_format,
332355
s3_endpoint_override=self.file_options.s3_endpoint_override,
356+
s3_url_style=self.file_options.s3_url_style or None,
333357
)
334358

335359
@staticmethod
@@ -341,6 +365,7 @@ def from_data_source(data_source: DataSource) -> "SavedDatasetStorage":
341365
if data_source.file_format
342366
else ParquetFormat(),
343367
s3_endpoint_override=data_source.s3_endpoint_override,
368+
s3_url_style=data_source.s3_url_style,
344369
)
345370

346371

@@ -349,24 +374,28 @@ class FileLoggingDestination(LoggingDestination):
349374

350375
path: str
351376
s3_endpoint_override: str
377+
s3_url_style: str
352378
partition_by: Optional[List[str]]
353379

354380
def __init__(
355381
self,
356382
*,
357383
path: str,
358384
s3_endpoint_override="",
385+
s3_url_style: str = "",
359386
partition_by: Optional[List[str]] = None,
360387
):
361388
self.path = path
362389
self.s3_endpoint_override = s3_endpoint_override
390+
self.s3_url_style = s3_url_style
363391
self.partition_by = partition_by
364392

365393
@classmethod
366394
def from_proto(cls, config_proto: LoggingConfigProto) -> "LoggingDestination":
367395
return FileLoggingDestination(
368396
path=config_proto.file_destination.path,
369397
s3_endpoint_override=config_proto.file_destination.s3_endpoint_override,
398+
s3_url_style=config_proto.file_destination.s3_url_style or "",
370399
partition_by=list(config_proto.file_destination.partition_by)
371400
if config_proto.file_destination.partition_by
372401
else None,
@@ -377,6 +406,7 @@ def to_proto(self) -> LoggingConfigProto:
377406
file_destination=LoggingConfigProto.FileDestination(
378407
path=self.path,
379408
s3_endpoint_override=self.s3_endpoint_override,
409+
s3_url_style=self.s3_url_style,
380410
partition_by=self.partition_by,
381411
)
382412
)
@@ -386,4 +416,5 @@ def to_data_source(self) -> DataSource:
386416
path=self.path,
387417
file_format=ParquetFormat(),
388418
s3_endpoint_override=self.s3_endpoint_override,
419+
s3_url_style=self.s3_url_style or None,
389420
)

‎sdk/python/feast/infra/offline_stores/ibis.py‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ def pull_latest_from_table_or_query_ibis(
5050
data_source_writer: Callable[[pyarrow.Table, DataSource, str], None],
5151
staging_location: Optional[str] = None,
5252
staging_location_endpoint_override: Optional[str] = None,
53+
staging_location_s3_url_style: Optional[str] = None,
5354
) -> RetrievalJob:
5455
fields = join_key_columns + feature_name_columns + [timestamp_field]
5556
if created_timestamp_column:
@@ -87,6 +88,7 @@ def pull_latest_from_table_or_query_ibis(
8788
data_source_writer=data_source_writer,
8889
staging_location=staging_location,
8990
staging_location_endpoint_override=staging_location_endpoint_override,
91+
staging_location_s3_url_style=staging_location_s3_url_style,
9092
repo_path=str(config.repo_path),
9193
)
9294

@@ -153,6 +155,7 @@ def get_historical_features_ibis(
153155
full_feature_names: bool = False,
154156
staging_location: Optional[str] = None,
155157
staging_location_endpoint_override: Optional[str] = None,
158+
staging_location_s3_url_style: Optional[str] = None,
156159
event_expire_timestamp_fn=None,
157160
) -> RetrievalJob:
158161
entity_schema = _get_entity_schema(
@@ -250,6 +253,7 @@ def read_fv(
250253
data_source_writer=data_source_writer,
251254
staging_location=staging_location,
252255
staging_location_endpoint_override=staging_location_endpoint_override,
256+
staging_location_s3_url_style=staging_location_s3_url_style,
253257
repo_path=str(config.repo_path),
254258
)
255259

@@ -267,6 +271,7 @@ def pull_all_from_table_or_query_ibis(
267271
end_date: Optional[datetime] = None,
268272
staging_location: Optional[str] = None,
269273
staging_location_endpoint_override: Optional[str] = None,
274+
staging_location_s3_url_style: Optional[str] = None,
270275
) -> RetrievalJob:
271276
timestamp_fields = [timestamp_field]
272277
if created_timestamp_column:
@@ -304,6 +309,7 @@ def pull_all_from_table_or_query_ibis(
304309
data_source_writer=data_source_writer,
305310
staging_location=staging_location,
306311
staging_location_endpoint_override=staging_location_endpoint_override,
312+
staging_location_s3_url_style=staging_location_s3_url_style,
307313
repo_path=str(config.repo_path),
308314
)
309315

@@ -487,6 +493,7 @@ def __init__(
487493
staging_location,
488494
staging_location_endpoint_override,
489495
repo_path,
496+
staging_location_s3_url_style=None,
490497
) -> None:
491498
super().__init__()
492499
self.table = table
@@ -498,6 +505,7 @@ def __init__(
498505
self.data_source_writer = data_source_writer
499506
self.staging_location = staging_location
500507
self.staging_location_endpoint_override = staging_location_endpoint_override
508+
self.staging_location_s3_url_style = staging_location_s3_url_style
501509
self.repo_path = repo_path
502510

503511
def _to_df_internal(self, timeout: Optional[int] = None) -> pd.DataFrame:
@@ -538,7 +546,11 @@ def supports_remote_storage_export(self) -> bool:
538546
def to_remote_storage(self) -> List[str]:
539547
path = self.staging_location + f"/{str(uuid.uuid4())}"
540548

541-
storage_options = {"AWS_ENDPOINT_URL": self.staging_location_endpoint_override}
549+
storage_options = {}
550+
if self.staging_location_endpoint_override:
551+
storage_options["AWS_ENDPOINT_URL"] = self.staging_location_endpoint_override
552+
if self.staging_location_s3_url_style:
553+
storage_options["AWS_S3_URL_STYLE"] = self.staging_location_s3_url_style
542554

543555
self.table.to_delta(path, storage_options=storage_options)
544556

0 commit comments

Comments
 (0)