[xanderbailey]

Working towards #3222

Rationale for this change

We need to be able to read encrypted manifest lists which means we need the Table to be able to construct an EncryptionManager.

• Table takes an optional kms_client, and Table.encryption_manager builds the manager from the table's current metadata.
• Snapshot.manifests takes an optional encryption_manager. When the snapshot has a key-id the manifest list is decrypted, and it raises if no manager is given.
• Scans and inspect pass the table's encryption manager through.
• The manager is passed around separately rather than wrapped into FileIO, matching iceberg-rust ([7/N] feat(encryption) [7/N] Read encrypted manifest list iceberg-rust#2453). <- this is the main design decision to make for this PR and is worth considering carefully. I've chosen to take the approach we used in rust because it actually simplifies things down the line a lot but very happy to hear arguments for the alternative.

Catalogs don't pass a KMS client yet, that comes in a follow-up.

Are these changes tested?

Yes. A manifest list encrypted by iceberg-rust is checked in under tests/table/encryption, and scanning and inspecting that table reads it. There's also a round trip of an encrypted manifest list with entries.

Are there any user-facing changes?

Table and the scans take a new optional argument. Nothing changes for unencrypted tables.

[xanderbailey]

@kevinjqliu would be interested on your take here for how this looks vs how Java does it. I'm pretty happy with following the same pattern we have in rust but curious for your thoughts.

[mbutrovich]

Thanks @xanderbailey, first pass.

[mbutrovich]

Could we build the manager once per metadata version instead of on every access? EncryptionManager caches unwrapped KEKs for an hour (_unwrap_kek), but each access here creates a new manager with an empty cache, so every manifest list read calls KeyManagementClient.unwrap_key again. I wrapped the test KMS in table_v3_encrypted with a call counter. Three scan().plan_files() calls made three unwrap_key calls. inspect.all_manifests() over five snapshots made five, because _generate_manifests_table reads the property once per snapshot.

Table.metadata is reassigned on commit and refresh, so caching the manager next to the metadata it came from keeps test_encryption_manager_follows_current_metadata passing:

if self._encryption_manager is None or self._encryption_manager[0] is not self.metadata:
    self._encryption_manager = (self.metadata, EncryptionManager.from_table_metadata(self._kms_client, self.metadata))
return self._encryption_manager[1]

Here self._encryption_manager starts as None in __init__. With that change, three scans followed by manifests(), entries(), and all_manifests() made one unwrap_key call.

[xanderbailey]

Ah this kind of thing might work, I was worried about keeping the em fresh as metadata changes so thought I'd do the expensive thing first but let me look into what you've suggested here. Thanks Matt

[mbutrovich]

Should the DataScan a few lines below (L401-L407) also get encryption_manager=self.tbl.encryption_manager? This line decrypts the manifest list, but that scan is built without the manager, so _plan_manifest_entries calls snapshot.manifests(io, None). On the head commit, table_v3_encrypted.inspect.partitions() raises ValueError: Snapshot 1 is encrypted, but no encryption manager was provided. With the argument added, it returns an empty table.

As I read it, this is the main cost of passing the manager separately, which bears on the design question in the description. Every caller of Snapshot.manifests and every scan constructor has to remember it, and a missed one fails only on encrypted tables. In iceberg-rust, ManifestListReader::new is pub(crate), and callers get a reader from Table::manifest_list_reader, which supplies the manager, so they can't leave it out.

[mbutrovich]

Could this also cover entries() and partitions()? Both read the manifest list through the paths this PR changes, and partitions() would have caught the missing manager above.

Suggested change

	def test_inspect_encrypted_table(table_v3_encrypted: Table) -> None:
	assert table_v3_encrypted.inspect.manifests().num_rows == 0
	assert table_v3_encrypted.inspect.all_manifests().num_rows == 0
	def test_inspect_encrypted_table(table_v3_encrypted: Table) -> None:
	assert table_v3_encrypted.inspect.manifests().num_rows == 0
	assert table_v3_encrypted.inspect.all_manifests().num_rows == 0
	assert table_v3_encrypted.inspect.entries().num_rows == 0
	assert table_v3_encrypted.inspect.partitions().num_rows == 0

[kevinjqliu]

Is there anything special about this test file (manifest-list-v3-encrypted.avro) that it needs to be copied over?

I'd prefer to use the integration test infra to generate the encrypted avro file, if its possible. This avoids us having to keep a test file in the repo an making packaging easier.

We use dev/provision.py to generate desired iceberg tables using spark and then those tables can be used in integration tests.