CTTY commented on code in PR #3474:
URL: https://github.com/apache/iceberg-rust/pull/3474#discussion_r4383405607
##########
crates/iceberg/src/transaction/append.rs:
##########
@@ -73,94 +85,59 @@ impl FastAppendAction {
self
}
- /// Collapse files sharing a path to their first occurrence, so a single
- /// manifest never references the same file twice. Always runs (unlike the
- /// `check_duplicate`-gated cross-snapshot check) since it is in-memory
only.
- fn dedupe_added_files(&self) -> Vec<DataFile> {
- let mut seen = HashSet::with_capacity(self.added_data_files.len());
- self.added_data_files
+ /// Checks the parts of this action that do not depend on the table.
+ fn validate(&self) -> Result<()> {
+ // TODO: Allowing snapshot property setup with no added data files is
a workaround.
+ // We should clean it up after all necessary actions are supported.
+ // For details, please refer to
https://github.com/apache/iceberg-rust/issues/1548
+ if self.added_data_files.is_empty() &&
self.snapshot_properties.is_empty() {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "No added data files or added snapshot properties found when
write a manifest file",
+ ));
+ }
+ if let Some(data_file) = self
+ .added_data_files
.iter()
- .filter(|data_file| seen.insert(data_file.file_path.as_str()))
- .cloned()
- .collect()
+ .find(|f| f.content_type() != DataContentType::Data)
+ {
+ return Err(invalid_data!(
+ "Only data content type is allowed for fast append, got {:?}
for {}",
+ data_file.content_type(),
+ data_file.file_path()
+ ));
+ }
+ Ok(())
}
}
#[async_trait]
impl TransactionAction for FastAppendAction {
- // TODO: replace with a persistent `SimpleSnapshotProducer` once snapshot
- // production is migrated (stateful transaction RFC, milestone 2).
- type State = ();
-
- fn new_state(&self) -> Self::State {}
-
- async fn commit(&self, _state: &mut (), table: &Table) ->
Result<ActionCommit> {
- let snapshot_producer = SnapshotProducer::new(
- table,
- self.commit_uuid.unwrap_or_else(Uuid::now_v7),
- self.snapshot_properties.clone(),
- self.dedupe_added_files(),
- vec![],
- );
-
- // validate added files
- snapshot_producer.validate_added_data_files()?;
-
- // Checks duplicate files
- if self.check_duplicate {
- snapshot_producer.validate_duplicate_files().await?;
- }
-
- snapshot_producer
- .commit(FastAppendOperation, DefaultManifestProcess)
- .await
- }
-
- async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {
- // Stateless for now: generated manifests are not yet tracked for
- // terminal cleanup (stateful transaction RFC, milestone 2/3).
+ type State = SimpleSnapshotProducer;
+
+ fn new_state(&self) -> SimpleSnapshotProducer {
+ SimpleSnapshotProducer::new(AppendIntent {
+ commit_uuid: self.commit_uuid,
+ snapshot_properties: self.snapshot_properties.clone(),
+ data_files: self.added_data_files.clone(),
+ })
}
-}
-
-struct FastAppendOperation;
-impl SnapshotProduceOperation for FastAppendOperation {
- fn operation(&self) -> Operation {
- Operation::Append
- }
-
- async fn delete_entries(
+ async fn commit(
&self,
- _snapshot_produce: &SnapshotProducer<'_>,
- ) -> Result<Vec<ManifestEntry>> {
- Ok(vec![])
+ state: &mut SimpleSnapshotProducer,
+ table: &Table,
+ ) -> Result<ActionCommit> {
+ self.validate()?;
Review Comment:
This technically still runs every attempt, we can fix it by having a gate
like `if attemp <=0 then validate` but I think that would be hacky and doesn't
add a lot of value
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
