Python: Fine-tune the API

python/pyiceberg/schema.py

@@ -75,9 +75,7 @@ def __str__(self):
  return "table {\n" + "\n".join([" " + str(field) for field in self.columns]) + "\n}"
 
  def __repr__(self):
- return (
- f"Schema(fields={repr(self.columns)}, schema_id={self.schema_id}, identifier_field_ids={self.identifier_field_ids})"
- )
+ return f"Schema({', '.join(repr(column) for column in self.columns)}, schema_id={self.schema_id}, identifier_field_ids={self.identifier_field_ids})"
 
  def __eq__(self, other) -> bool:
  if not other:

python/pyiceberg/table/metadata.py

@@ -46,6 +46,8 @@
 CURRENT_SNAPSHOT_ID = "current_snapshot_id"
 CURRENT_SCHEMA_ID = "current_schema_id"
 SCHEMAS = "schemas"
+DEFAULT_SPEC_ID = "default_spec_id"
+PARTITION_SPEC = "partition_spec"
 PARTITION_SPECS = "partition_specs"
 SORT_ORDERS = "sort_orders"
 REFS = "refs"
@@ -261,8 +263,10 @@ def construct_partition_specs(cls, data: Dict[str, Any]) -> Dict[str, Any]:
  The TableMetadata with the partition_specs set, if not provided
  """
  if not data.get(PARTITION_SPECS):
- fields = data["partition_spec"]
- data[PARTITION_SPECS] = [PartitionSpec(spec_id=INITIAL_SPEC_ID, fields=fields)]
+ fields = data[PARTITION_SPEC]
+ migrated_spec = PartitionSpec(*fields)
+ data[PARTITION_SPECS] = [migrated_spec]
+ data[DEFAULT_SPEC_ID] = migrated_spec.spec_id
  else:
  check_partition_specs(data)
 

python/pyiceberg/table/partitioning.py

@@ -29,7 +29,7 @@
 from pyiceberg.transforms import Transform
 from pyiceberg.utils.iceberg_base_model import IcebergBaseModel
 
-INITIAL_SPEC_ID = 0
+INITIAL_PARTITION_SPEC_ID = 0
 _PARTITION_DATA_ID_START: int = 1000
 
 
@@ -82,19 +82,16 @@ class PartitionSpec(IcebergBaseModel):
  fields(List[PartitionField): list of partition fields to produce partition values
  """
 
- spec_id: int = Field(alias="spec-id")
- fields: Tuple[PartitionField, ...] = Field(default_factory=tuple)
+ spec_id: int = Field(alias="spec-id", default=INITIAL_PARTITION_SPEC_ID)
+ fields: Tuple[PartitionField, ...] = Field(alias="fields", default_factory=tuple)
 
  def __init__(
  self,
- spec_id: Optional[int] = None,
- fields: Optional[Tuple[PartitionField, ...]] = None,
+ *fields: PartitionField,
  **data: Any,
  ):
- if spec_id is not None:
- data["spec-id"] = spec_id
- if fields is not None:
- data["fields"] = fields
+ if fields:
+ data["fields"] = tuple(fields)
  super().__init__(**data)
 
  def __eq__(self, other: Any) -> bool:
@@ -121,6 +118,10 @@ def __str__(self):
  result_str += "]"
  return result_str
 
+ def __repr__(self) -> str:
+ fields = f"{', '.join(repr(column) for column in self.fields)}, " if self.fields else ""
+ return f"PartitionSpec({fields}spec_id={self.spec_id})"
+
  def is_unpartitioned(self) -> bool:
  return not self.fields
 
@@ -178,4 +179,4 @@ def assign_fresh_partition_spec_ids(spec: PartitionSpec, old_schema: Schema, fre
  transform=field.transform,
  )
  )
- return PartitionSpec(INITIAL_SPEC_ID, fields=tuple(partition_fields))
+ return PartitionSpec(*partition_fields, spec_id=INITIAL_PARTITION_SPEC_ID)

python/pyiceberg/table/sorting.py

@@ -104,6 +104,9 @@ def __str__(self):
  return f"{self.transform}({self.source_id}) {self.direction} {self.null_order}"
 
 
+INITIAL_SORT_ORDER_ID = 1
+
+
 class SortOrder(IcebergBaseModel):
  """Describes how the data is sorted within the table
 
@@ -112,20 +115,18 @@ class SortOrder(IcebergBaseModel):
  The order of the sort fields within the list defines the order in which the sort is applied to the data.
 
  Args:
- order_id (int): The id of the sort-order. To keep track of historical sorting
+ order_id (int): An unique id of the sort-orderof a table.
  fields (List[SortField]): The fields how the table is sorted
  """
 
- def __init__(self, order_id: Optional[int] = None, *fields: SortField, **data: Any):
- if order_id is not None:
- data["order-id"] = order_id
+ order_id: int = Field(alias="order-id", default=INITIAL_SORT_ORDER_ID)
+ fields: List[SortField] = Field(default_factory=list)
+
+ def __init__(self, *fields: SortField, **data: Any):
  if fields:
  data["fields"] = fields
  super().__init__(**data)
 
- order_id: int = Field(alias="order-id")
- fields: List[SortField] = Field(default_factory=list)
-
  @property
  def is_unsorted(self) -> bool:
  return len(self.fields) == 0
@@ -137,10 +138,13 @@ def __str__(self) -> str:
  result_str += "]"
  return result_str
 
+ def __repr__(self):
+ fields = f"{', '.join(repr(column) for column in self.fields)}, " if self.fields else ""
+ return f"SortOrder({fields}order_id={self.order_id})"
+
 
 UNSORTED_SORT_ORDER_ID = 0
 UNSORTED_SORT_ORDER = SortOrder(order_id=UNSORTED_SORT_ORDER_ID)
-INITIAL_SORT_ORDER_ID = 1
 
 
 def assign_fresh_sort_order_ids(sort_order: SortOrder, old_schema: Schema, fresh_schema: Schema) -> SortOrder:
@@ -164,7 +168,4 @@ def assign_fresh_sort_order_ids(sort_order: SortOrder, old_schema: Schema, fresh
  )
  )
 
- return SortOrder(
- INITIAL_SORT_ORDER_ID,
- *fresh_fields,
- )
+ return SortOrder(*fresh_fields, order_id=INITIAL_SORT_ORDER_ID)

python/pyproject.toml

@@ -51,9 +51,9 @@ pyyaml = "^6.0.0"
 pydantic = "^1.10.2"
 fsspec = "2022.8.2"
 
-pyarrow = { version = "^9.0.0", optional = true }
+zstandard = "^0.18.0"
 
-zstandard = { version = "^0.18.0", optional = true }
+pyarrow = { version = "^9.0.0", optional = true }
 
 python-snappy = { version = "^0.6.1", optional = true }
 
@@ -80,7 +80,6 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry.extras]
 pyarrow = ["pyarrow"]
 snappy = ["python-snappy"]
-python-snappy = ["zstandard"]
 hive = ["thrift"]
 s3fs = ["s3fs"]
 

python/tests/catalog/test_base.py

@@ -40,9 +40,9 @@
 )
 from pyiceberg.schema import Schema
 from pyiceberg.table import Table
-from pyiceberg.table.metadata import INITIAL_SPEC_ID
-from pyiceberg.table.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
+from pyiceberg.table.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionField, PartitionSpec
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
+from pyiceberg.transforms import IdentityTransform
 from pyiceberg.typedef import EMPTY_DICT
 from tests.table.test_metadata import EXAMPLE_TABLE_METADATA_V1
 
@@ -186,7 +186,7 @@ def update_namespace_properties(
 TEST_TABLE_NAME = "my_table"
 TEST_TABLE_SCHEMA = Schema(schema_id=1)
 TEST_TABLE_LOCATION = "protocol://some/location"
-TEST_TABLE_PARTITION_SPEC = PartitionSpec(spec_id=INITIAL_SPEC_ID, fields=())
+TEST_TABLE_PARTITION_SPEC = PartitionSpec(PartitionField(name="x", transform=IdentityTransform(), source_id=1, field_id=1000))
 TEST_TABLE_PROPERTIES = {"key1": "value1", "key2": "value2"}
 NO_SUCH_TABLE_ERROR = "Table does not exist: \\('com', 'organization', 'department', 'my_table'\\)"
 TABLE_ALREADY_EXISTS_ERROR = "Table already exists: \\('com', 'organization', 'department', 'my_table'\\)"
@@ -200,7 +200,7 @@ def given_catalog_has_a_table(catalog: InMemoryCatalog) -> Table:
  identifier=TEST_TABLE_IDENTIFIER,
  schema=TEST_TABLE_SCHEMA,
  location=TEST_TABLE_LOCATION,
- partition_spec=TEST_TABLE_PARTITION_SPEC,
+ partition_spec=UNPARTITIONED_PARTITION_SPEC,
  properties=TEST_TABLE_PROPERTIES,
  )
 

python/tests/catalog/test_hive.py

@@ -77,7 +77,8 @@
 @pytest.fixture
 def hive_table(tmp_path_factory, example_table_metadata_v2: Dict[str, Any]) -> HiveTable:
  metadata_path = str(tmp_path_factory.mktemp("metadata") / f"{uuid.uuid4()}.metadata.json")
- ToOutputFile.table_metadata(TableMetadataV2(**example_table_metadata_v2), LocalFileIO().new_output(str(metadata_path)), True)
+ metadata = TableMetadataV2(**example_table_metadata_v2)
+ ToOutputFile.table_metadata(metadata, LocalFileIO().new_output(str(metadata_path)), True)
 
  return HiveTable(
  tableName="new_tabl2e",
@@ -267,7 +268,7 @@ def test_create_table(table_schema_simple: Schema, hive_database: HiveDatabase,
 
  assert "database/table" in metadata.location
 
-assert metadata == TableMetadataV2(
+expected = TableMetadataV2(
  location=metadata.location,
  table_uuid=metadata.table_uuid,
  last_updated_ms=metadata.last_updated_ms,
@@ -282,10 +283,10 @@ def test_create_table(table_schema_simple: Schema, hive_database: HiveDatabase,
  )
  ],
  current_schema_id=0,
- partition_specs=[PartitionSpec(spec_id=0)],
- default_spec_id=0,
  last_partition_id=1000,
  properties={"owner": "javaberg"},
+ partition_specs=[PartitionSpec()],
+ default_spec_id=0,
  current_snapshot_id=None,
  snapshots=[],
  snapshot_log=[],
@@ -297,6 +298,8 @@ def test_create_table(table_schema_simple: Schema, hive_database: HiveDatabase,
  last_sequence_number=0,
  )
 
+ assert metadata.dict() == expected.dict()
+
 
 def test_load_table(hive_table: HiveTable):
  catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
@@ -307,8 +310,7 @@ def test_load_table(hive_table: HiveTable):
 
  catalog._client.__enter__().get_table.assert_called_with(dbname="default", tbl_name="new_tabl2e")
 
- assert table.identifier == ("default", "new_tabl2e")
- assert table.metadata == TableMetadataV2(
+ expected = TableMetadataV2(
  location="s3://bucket/test/location",
  table_uuid=uuid.UUID("9c12d441-03fe-4693-9a96-a0705ddf69c1"),
  last_updated_ms=1602638573590,
@@ -329,9 +331,7 @@ def test_load_table(hive_table: HiveTable):
  ],
  current_schema_id=1,
  partition_specs=[
- PartitionSpec(
- spec_id=0, fields=(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"),)
- )
+ PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"), spec_id=0)
  ],
  default_spec_id=0,
  last_partition_id=1000,
@@ -364,7 +364,6 @@ def test_load_table(hive_table: HiveTable):
  metadata_log=[MetadataLogEntry(metadata_file="s3://bucket/.../v1.json", timestamp_ms=1515100)],
  sort_orders=[
  SortOrder(
- 3,
  SortField(
  source_id=2, transform=IdentityTransform(), direction=SortDirection.ASC, null_order=NullOrder.NULLS_FIRST
  ),
@@ -374,6 +373,7 @@ def test_load_table(hive_table: HiveTable):
  direction=SortDirection.DESC,
  null_order=NullOrder.NULLS_LAST,
  ),
+ order_id=3,
  )
  ],
  default_sort_order_id=3,
@@ -397,6 +397,9 @@ def test_load_table(hive_table: HiveTable):
  last_sequence_number=34,
  )
 
+ assert table.identifier == ("default", "new_tabl2e")
+ assert expected == table.metadata
+
 
 def test_rename_table_from_does_not_exists():
  catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)

python/tests/catalog/test_rest.py

@@ -344,8 +344,8 @@ def test_load_table_200(rest_mock: Mocker):
  },
  status_code=200,
  )
-table = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).load_table(("fokko", "table"))
-assert table == Table(
+actual = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).load_table(("fokko", "table"))
+expected = Table(
  identifier=("rest", "fokko", "table"),
  metadata_location="s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json",
  metadata=TableMetadataV1(
@@ -362,7 +362,6 @@ def test_load_table_200(rest_mock: Mocker):
  )
  ],
  current_schema_id=0,
- partition_specs=[PartitionSpec(spec_id=0, fields=())],
  default_spec_id=0,
  last_partition_id=999,
  properties={"owner": "bryan", "write.metadata.compression-codec": "gzip"},
@@ -422,6 +421,7 @@ def test_load_table_200(rest_mock: Mocker):
  ),
  config={"client.factory": "io.tabular.iceberg.catalog.TabularAwsClientFactory", "region": "us-west-2"},
  )
+ assert actual == expected
 
 
 def test_load_table_404(rest_mock: Mocker):
@@ -496,7 +496,6 @@ def test_create_table_200(rest_mock: Mocker, table_schema_simple: Schema):
  ],
  "partition-spec": [],
  "default-spec-id": 0,
- "partition-specs": [{"spec-id": 0, "fields": []}],
  "last-partition-id": 999,
  "default-sort-order-id": 0,
  "sort-orders": [{"order-id": 0, "fields": []}],
@@ -524,9 +523,9 @@ def test_create_table_200(rest_mock: Mocker, table_schema_simple: Schema):
  schema=table_schema_simple,
  location=None,
  partition_spec=PartitionSpec(
-spec_id=1, fields=(PartitionField(source_id=1, field_id=1000, transform=TruncateTransform(width=3), name="id"),)
+ PartitionField(source_id=1, field_id=1000, transform=TruncateTransform(width=3), name="id"), spec_id=1
  ),
- sort_order=SortOrder(1, SortField(source_id=2, transform=IdentityTransform())),
+ sort_order=SortOrder(SortField(source_id=2, transform=IdentityTransform())),
  properties={"owner": "fokko"},
  )
  assert table == Table(
@@ -547,7 +546,6 @@ def test_create_table_200(rest_mock: Mocker, table_schema_simple: Schema):
  )
  ],
  current_schema_id=0,
- partition_specs=[PartitionSpec(spec_id=0, fields=())],
  default_spec_id=0,
  last_partition_id=999,
  properties={
@@ -595,10 +593,9 @@ def test_create_table_409(rest_mock, table_schema_simple: Schema):
  schema=table_schema_simple,
  location=None,
  partition_spec=PartitionSpec(
- spec_id=1,
- fields=(PartitionField(source_id=1, field_id=1000, transform=TruncateTransform(width=3), name="id"),),
+ PartitionField(source_id=1, field_id=1000, transform=TruncateTransform(width=3), name="id")
  ),
- sort_order=SortOrder(1, SortField(source_id=2, transform=IdentityTransform())),
+ sort_order=SortOrder(SortField(source_id=2, transform=IdentityTransform())),
  properties={"owner": "fokko"},
  )
  assert "Table already exists" in str(e.value)

python/tests/table/test_init.py

@@ -82,13 +82,13 @@ def test_schemas(table):
 
 def test_spec(table):
  assert table.spec() == PartitionSpec(
-spec_id=0, fields=(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"),)
+ PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"), spec_id=0
  )
 
 
 def test_specs(table):
  assert table.specs() == {
- 0: PartitionSpec(spec_id=0, fields=(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"),))
+ 0: PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"), spec_id=0)
  }