iceberg_loader: Simplify batch_metadata

Author: fordN

src/amp/loaders/implementations/iceberg_loader.py

@@ -406,45 +406,13 @@ def _perform_load_operation(self, iceberg_table: IcebergTable, arrow_table: pa.T
 
     def _get_loader_batch_metadata(self, batch: pa.RecordBatch, duration: float, **kwargs) -> Dict[str, Any]:
         """Get Iceberg-specific metadata for batch operation"""
-        metadata = {'namespace': self.config.namespace}
-
-        # Add partition columns if available
-        table_name = kwargs.get('table_name')
-        if table_name and self._table_exists(table_name):
-            try:
-                table_info = self.get_table_info(table_name)
-                metadata['partition_columns'] = table_info.get('partition_columns', [])
-            except Exception:
-                metadata['partition_columns'] = []
-        else:
-            # For new tables, get partition fields from partition_spec if available
-            metadata['partition_columns'] = []
-
-        return metadata
+        return {'namespace': self.config.namespace}
 
     def _get_loader_table_metadata(
-        self, table: pa.Table, duration: float, batch_count: int, **kwargs
+            self, table: pa.Table, duration: float, batch_count: int, **kwargs
     ) -> Dict[str, Any]:
         """Get Iceberg-specific metadata for table operation"""
-        metadata = {'namespace': self.config.namespace}
-
-        # Add partition columns if available
-        table_name = kwargs.get('table_name')
-        if table_name and self._table_exists(table_name):
-            try:
-                table_info = self.get_table_info(table_name)
-                metadata['partition_columns'] = table_info.get('partition_columns', [])
-            except Exception:
-                metadata['partition_columns'] = []
-        else:
-            # For new tables, get partition fields from partition_spec if available
-            metadata['partition_columns'] = []
-            if self.config.partition_spec and hasattr(self.config.partition_spec, 'fields'):
-                # partition_spec.fields contains partition field definitions
-                # We'll extract them during table creation
-                metadata['partition_columns'] = []  # Will be populated after table creation
-
-        return metadata
+        return {'namespace': self.config.namespace}
 
     def _table_exists(self, table_name: str) -> bool:
         """Check if a table exists"""