Sync S3 and IcebergRestS3 bucket_style option

Author: dolfinus

.env.docker.test

@@ -57,7 +57,7 @@ TEST_ICEBERG_METASTORE_USERNAME=syncmaster
 TEST_ICEBERG_METASTORE_PASSWORD=123UsedForTestOnly@!
 TEST_ICEBERG_S3_WAREHOUSE_PATH=/data
 TEST_ICEBERG_S3_REGION=us-east-1
-TEST_ICEBERG_S3_PATH_STYLE_ACCESS=True
+TEST_ICEBERG_S3_BUCKET_STYLE=path
 
 TEST_HDFS_HOST=test-hive
 TEST_HDFS_WEBHDFS_PORT=9870

.env.local.test

@@ -57,7 +57,7 @@ export TEST_ICEBERG_METASTORE_USERNAME=syncmaster
 export TEST_ICEBERG_METASTORE_PASSWORD=123UsedForTestOnly@!
 export TEST_ICEBERG_S3_WAREHOUSE_PATH=/data
 export TEST_ICEBERG_S3_REGION=us-east-1
-export TEST_ICEBERG_S3_PATH_STYLE_ACCESS=True
+export TEST_ICEBERG_S3_BUCKET_STYLE=path
 
 export TEST_HDFS_HOST=test-hive
 export TEST_HDFS_WEBHDFS_PORT=9870

syncmaster/dto/connections.py

@@ -79,14 +79,14 @@ class IcebergRESTCatalogS3ConnectionDTO(ConnectionDTO):
     s3_warehouse_path: str
     s3_host: str
     s3_bucket: str
+    s3_bucket_style: str
     s3_region: str
     s3_access_key: str
     s3_secret_key: str
     metastore_username: str
     metastore_password: str
     s3_port: int | None = None
     s3_protocol: str = "https"
-    s3_path_style_access: bool = False
     type: ClassVar[str] = "iceberg_rest_s3"
 
 
@@ -105,6 +105,7 @@ class S3ConnectionDTO(ConnectionDTO):
     access_key: str
     secret_key: str
     bucket: str
+    bucket_style: str
     additional_params: dict
     region: str | None = None
     protocol: str = "https"

syncmaster/schemas/v1/connections/iceberg.py

@@ -25,7 +25,7 @@ class CreateIcebergRESTCatalogS3ConnectionDataSchema(BaseModel):
     s3_protocol: Literal["http", "https"] = "https"
     s3_bucket: str
     s3_region: str
-    s3_path_style_access: bool = False
+    s3_bucket_style: Literal["domain", "path"] = "path"
 
 
 class ReadIcebergRESTCatalogS3ConnectionDataSchema(BaseModel):
@@ -36,7 +36,7 @@ class ReadIcebergRESTCatalogS3ConnectionDataSchema(BaseModel):
     s3_protocol: Literal["http", "https"] = "https"
     s3_bucket: str
     s3_region: str
-    s3_path_style_access: bool = False
+    s3_bucket_style: Literal["domain", "path"] = "path"
 
 
 class CreateIcebergConnectionSchema(CreateConnectionBaseSchema):

syncmaster/worker/handlers/db/iceberg.py

@@ -44,7 +44,7 @@ def connect(self, spark: SparkSession):
                 port=self.connection_dto.s3_port,
                 protocol=self.connection_dto.s3_protocol,
                 bucket=self.connection_dto.s3_bucket,
-                path_style_access=self.connection_dto.s3_path_style_access,
+                path_style_access=self.connection_dto.s3_bucket_style == "path",
                 region=self.connection_dto.s3_region,
                 access_key=self.connection_dto.s3_access_key,
                 secret_key=self.connection_dto.s3_secret_key,
@@ -53,7 +53,7 @@ def connect(self, spark: SparkSession):
 
     @slot
     def read(self) -> DataFrame:
-        table = f"{self.transfer_dto.catalog_name}.{self.transfer_dto.table_name}"
+        table = ".".join([self.transfer_dto.catalog_name, self.transfer_dto.table_name])
         self.connection.spark.catalog.refreshTable(table)
         return super().read()
 

syncmaster/worker/handlers/file/s3.py

@@ -28,6 +28,7 @@ def connect(self, spark: SparkSession):
             access_key=self.connection_dto.access_key,
             secret_key=self.connection_dto.secret_key,
             bucket=self.connection_dto.bucket,
+            path_style_access=self.connection_dto.bucket_style == "path",
             protocol=self.connection_dto.protocol,
             region=self.connection_dto.region,
             extra=self.connection_dto.additional_params,

tests/settings.py

@@ -56,7 +56,7 @@ class TestSettings(BaseSettings):
     TEST_ICEBERG_METASTORE_PASSWORD: str
     TEST_ICEBERG_S3_WAREHOUSE_PATH: str
     TEST_ICEBERG_S3_REGION: str
-    TEST_ICEBERG_S3_PATH_STYLE_ACCESS: bool = True
+    TEST_ICEBERG_S3_BUCKET_STYLE: bool = True
 
     TEST_HDFS_HOST: str
     TEST_HDFS_WEBHDFS_PORT: int

tests/test_integration/test_run_transfer/connection_fixtures/iceberg_fixtures.py

@@ -30,7 +30,7 @@ def iceberg_rest_s3_for_conftest(test_settings: TestSettings) -> IcebergRESTCata
         s3_protocol=test_settings.TEST_S3_PROTOCOL,
         s3_bucket=test_settings.TEST_S3_BUCKET,
         s3_region=test_settings.TEST_ICEBERG_S3_REGION,
-        s3_path_style_access=test_settings.TEST_ICEBERG_S3_PATH_STYLE_ACCESS,
+        s3_bucket_style=test_settings.TEST_ICEBERG_S3_BUCKET_STYLE,
         s3_access_key=test_settings.TEST_S3_ACCESS_KEY,
         s3_secret_key=test_settings.TEST_S3_SECRET_KEY,
         metastore_username=test_settings.TEST_ICEBERG_METASTORE_USERNAME,
@@ -51,7 +51,7 @@ def iceberg_rest_s3_for_worker(test_settings: TestSettings) -> IcebergRESTCatalo
         s3_protocol=test_settings.TEST_S3_PROTOCOL,
         s3_bucket=test_settings.TEST_S3_BUCKET,
         s3_region=test_settings.TEST_ICEBERG_S3_REGION,
-        s3_path_style_access=test_settings.TEST_ICEBERG_S3_PATH_STYLE_ACCESS,
+        s3_bucket_style=test_settings.TEST_ICEBERG_S3_BUCKET_STYLE,
         s3_access_key=test_settings.TEST_S3_ACCESS_KEY,
         s3_secret_key=test_settings.TEST_S3_SECRET_KEY,
         metastore_username=test_settings.TEST_ICEBERG_METASTORE_USERNAME,
@@ -87,7 +87,7 @@ def prepare_iceberg_rest_s3(
             port=iceberg.s3_port,
             protocol=iceberg.s3_protocol,
             bucket=iceberg.s3_bucket,
-            path_style_access=iceberg.s3_path_style_access,
+            path_style_access=iceberg.s3_bucket_style == "path",
             region=iceberg.s3_region,
             access_key=iceberg.s3_access_key,
             secret_key=iceberg.s3_secret_key,
@@ -134,7 +134,7 @@ async def iceberg_rest_s3_connection(
             s3_protocol=iceberg.s3_protocol,
             s3_bucket=iceberg.s3_bucket,
             s3_region=iceberg.s3_region,
-            s3_path_style_access=iceberg.s3_path_style_access,
+            s3_bucket_style=iceberg.s3_bucket_style,
         ),
         group_id=group.id,
     )

tests/test_unit/test_connections/test_db_connection/test_create_iceberg_connection.py

@@ -36,7 +36,7 @@ async def test_developer_plus_can_create_iceberg_rest_s3_connection(
                 "s3_port": 9010,
                 "s3_bucket": "some_bucket",
                 "s3_region": "us-east-1",
-                "s3_path_style_access": True,
+                "s3_bucket_style": True,
             },
             "auth_data": {
                 "type": "iceberg_rest_basic_s3_basic",
@@ -79,7 +79,7 @@ async def test_developer_plus_can_create_iceberg_rest_s3_connection(
             "s3_port": connection.data["s3_port"],
             "s3_bucket": connection.data["s3_bucket"],
             "s3_region": connection.data["s3_region"],
-            "s3_path_style_access": connection.data["s3_path_style_access"],
+            "s3_bucket_style": connection.data["s3_bucket_style"],
         },
         "auth_data": {
             "type": decrypted["type"],

tests/test_unit/test_connections/test_db_connection/test_update_iceberg_connection.py

@@ -20,7 +20,7 @@
                 "s3_port": 9010,
                 "s3_bucket": "some_bucket",
                 "s3_region": "us-east-1",
-                "s3_path_style_access": True,
+                "s3_bucket_style": True,
             },
             {
                 "type": "iceberg_rest_basic_s3_basic",
@@ -79,7 +79,7 @@ async def test_developer_plus_can_update_iceberg_rest_s3_connection(
             "s3_port": None,
             "s3_bucket": "new_bucket",
             "s3_region": "us-east-2",
-            "s3_path_style_access": False,
+            "s3_bucket_style": False,
         },
         "auth_data": {
             "type": group_connection.credentials.value["type"],