This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new fc1e09f0 feat(core): add append_only writes for append-only tables
(#588)
fc1e09f0 is described below
commit fc1e09f0b6e02390af01becce78b1f8c03a1285b
Author: Sourav <[email protected]>
AuthorDate: Wed Oct 28 07:57:31 2026 +0530
feat(core): add append_only writes for append-only tables (#588)
Adds append_only support for append-only tables in hudi-core. So the writer
creates a Parquet base file, writes the matching commit metadata, and rejects
partitioned tables for now. The result now exposes the actual commit-relative
path, so callers can find the commit file in both timeline layouts.
---
crates/core/src/error.rs | 3 +
crates/core/src/table/mod.rs | 71 +++++
crates/core/src/table/writer.rs | 330 +++++++++++++++++++++++
crates/core/tests/table_bulk_insert_tests.rs | 386 +++++++++++++++++++++++++++
4 files changed, 790 insertions(+)
diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs
index 610deb5e..c49e7f3e 100644
--- a/crates/core/src/error.rs
+++ b/crates/core/src/error.rs
@@ -94,6 +94,9 @@ pub enum CoreError {
#[error("{0}")]
Unsupported(String),
+ #[error("Write error: {0}")]
+ Write(String),
+
#[error(transparent)]
Utf8Error(#[from] std::str::Utf8Error),
}
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index 76637d61..359f29b5 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -94,8 +94,10 @@ pub(crate) mod fs_view;
mod listing;
pub mod partition;
mod validation;
+mod writer;
pub use crate::config::read_options::{QueryType, ReadOptions};
+pub use writer::AppendResult;
use crate::Result;
use crate::config::HudiConfigs;
@@ -120,6 +122,7 @@ use crate::table::builder::TableBuilder;
use crate::table::file_pruner::FilePruner;
use crate::table::fs_view::FileSystemView;
use crate::table::partition::{PartitionPruner, project_partition_schema};
+use crate::table::writer::AppendOnlyWriter;
use crate::timeline::util::format_timestamp;
use crate::timeline::{EARLIEST_START_TIMESTAMP, Timeline};
use arrow::record_batch::RecordBatch;
@@ -1234,6 +1237,74 @@ impl Table {
Ok(Box::pin(combined_stream))
}
+ /// Append record batches to an append-only table as one new base file.
+ ///
+ /// Intentionally narrow: it only supports unpartitioned append-only
tables,
+ /// and all batches must share one schema matching the table's.
+ pub async fn append_only(
+ &mut self,
+ batches: impl IntoIterator<Item = RecordBatch>,
+ ) -> Result<AppendResult> {
+ let batches: Vec<RecordBatch> = batches.into_iter().collect();
+ let Some(first) = batches.first() else {
+ return Err(CoreError::Write(
+ "append requires at least one RecordBatch".to_string(),
+ ));
+ };
+ let schema = first.schema();
+ if batches.iter().any(|batch| batch.schema() != schema) {
+ return Err(CoreError::Write(
+ "All RecordBatches in append must share the same
schema".to_string(),
+ ));
+ }
+ self.ensure_append_schema_matches_table(schema.as_ref())
+ .await?;
+ let batch = arrow::compute::concat_batches(&schema, &batches)?;
+ if batch.num_rows() == 0 {
+ return Err(CoreError::Write(
+ "append requires at least one row".to_string(),
+ ));
+ }
+ let result = AppendOnlyWriter::new(self).write(batch).await?;
+ // The commit is durable; a failed refresh must not turn it into an
error.
+ if let Ok(timeline) =
+ Timeline::new_from_storage(self.hudi_configs.clone(),
self.storage_options.clone())
+ .await
+ {
+ self.timeline = timeline;
+ }
+ if let Ok(file_system_view) =
+ FileSystemView::new(self.hudi_configs.clone(),
self.storage_options.clone()).await
+ {
+ self.file_system_view = file_system_view;
+ }
+ self.cached_metadata_table = Arc::new(OnceCell::new());
+ self.cached_estimator = Arc::new(OnceCell::new());
+ Ok(result)
+ }
+
+ async fn ensure_append_schema_matches_table(&self, batch_schema: &Schema)
-> Result<()> {
+ if self.timeline.completed_commits.is_empty() {
+ return Ok(());
+ }
+ let table_schema = self.get_schema().await?;
+ let fields = |schema: &Schema| {
+ schema
+ .fields()
+ .iter()
+ .map(|f| (f.name().clone(), f.data_type().clone()))
+ .collect::<Vec<_>>()
+ };
+ if fields(&table_schema) == fields(batch_schema) {
+ return Ok(());
+ }
+ Err(CoreError::Write(format!(
+ "append batch schema does not match table schema (table
fields={:?}, batch fields={:?})",
+ fields(&table_schema),
+ fields(batch_schema),
+ )))
+ }
+
/// Compute estimated table-level statistics for scan planning.
///
/// Returns `(estimated_num_rows, estimated_total_byte_size)` derived from
diff --git a/crates/core/src/table/writer.rs b/crates/core/src/table/writer.rs
new file mode 100644
index 00000000..c206d113
--- /dev/null
+++ b/crates/core/src/table/writer.rs
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::str::FromStr;
+use std::sync::atomic::{AtomicI64, Ordering};
+
+use apache_avro::Writer as AvroWriter;
+use apache_avro::schema::AvroSchema;
+use arrow::record_batch::RecordBatch;
+use bytes::Bytes;
+use chrono::Utc;
+use object_store::path::Path;
+use object_store::{ObjectStoreExt, PutPayload};
+use parquet::arrow::ArrowWriter;
+
+use crate::Result;
+use crate::config::table::HudiTableConfig::{
+ MetadataTablePartitions, PopulatesMetaFields, RecordMergeStrategy,
TableVersion,
+ TimelineLayoutVersion, TimelinePath,
+};
+use crate::error::CoreError;
+use crate::merge::RecordMergeStrategyValue;
+use crate::metadata::HUDI_METADATA_DIR;
+use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat};
+use crate::storage::error::StorageError;
+use crate::storage::util::join_url_segments;
+use crate::table::Table;
+use crate::timeline::instant::Instant;
+
+/// Result of an append write.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AppendResult {
+ pub instant: String,
+ pub commit_relative_path: String,
+ pub base_file_path: String,
+ pub num_rows: usize,
+}
+
+/// Minimal writer for append-only tables.
+#[derive(Clone, Debug)]
+pub(crate) struct AppendOnlyWriter {
+ table: Table,
+}
+
+static LAST_EPOCH_MILLIS: AtomicI64 = AtomicI64::new(0);
+
+impl AppendOnlyWriter {
+ pub(crate) fn new(table: &Table) -> Self {
+ Self {
+ table: table.clone(),
+ }
+ }
+
+ pub(crate) async fn write(&self, batch: RecordBatch) ->
Result<AppendResult> {
+ self.ensure_append_only()?;
+ self.ensure_unpartitioned()?;
+ self.ensure_no_meta_fields()?;
+ self.ensure_no_metadata_table().await?;
+
+ let request_instant = Self::current_timestamp();
+ let completion_instant = request_instant.clone();
+ let layout_two = self.is_layout_two();
+ let instant_name = if layout_two {
+ format!("{request_instant}_{completion_instant}.commit")
+ } else {
+ format!("{request_instant}.commit")
+ };
+ let instant = Instant::from_str(&instant_name)?;
+ let timeline_dir = self.timeline_dir();
+
+ let file_id = format!("bulk-insert-{request_instant}");
+ let write_token = "0-0-0";
+ let base_file_name =
format!("{file_id}_{write_token}_{request_instant}.parquet");
+ let base_file_path = base_file_name.clone();
+
+ let file_bytes = self.write_parquet_bytes(&batch)?;
+ let file_size = file_bytes.len() as i64;
+
+ self.put_file(&base_file_path, file_bytes).await?;
+
+ let commit_metadata = self.build_commit_metadata(
+ &file_id,
+ &base_file_path,
+ batch.num_rows() as i64,
+ file_size,
+ );
+ let commit_relative_path =
instant.relative_path_with_base(&timeline_dir)?;
+ if let Err(error) = self
+ .write_commit_file(&commit_relative_path, &commit_metadata)
+ .await
+ {
+ let _ = self.delete_file(&base_file_path).await;
+ return Err(error);
+ }
+
+ Ok(AppendResult {
+ instant: request_instant,
+ commit_relative_path,
+ base_file_path,
+ num_rows: batch.num_rows(),
+ })
+ }
+
+ fn ensure_append_only(&self) -> Result<()> {
+ let strategy: String = self
+ .table
+ .hudi_configs
+ .get_or_default(RecordMergeStrategy)
+ .into();
+
+ if RecordMergeStrategyValue::from_str(&strategy)? !=
RecordMergeStrategyValue::AppendOnly {
+ return Err(CoreError::Unsupported(
+ "append_only requires
hoodie.record.merge.mode=append_only".to_string(),
+ ));
+ }
+
+ Ok(())
+ }
+
+ fn ensure_unpartitioned(&self) -> Result<()> {
+ let partition_fields: Vec<String> = self
+ .table
+ .hudi_configs
+
.get_or_default(crate::config::table::HudiTableConfig::PartitionFields)
+ .into();
+
+ if !partition_fields.is_empty() {
+ return Err(CoreError::Unsupported(
+ "append_only currently supports only unpartitioned append-only
tables".to_string(),
+ ));
+ }
+
+ Ok(())
+ }
+
+ /// The written base file carries only the batch's columns, so a table that
+ /// populates the `_hoodie_*` meta fields would get a file without them.
+ fn ensure_no_meta_fields(&self) -> Result<()> {
+ let populates_meta_fields: bool = self
+ .table
+ .hudi_configs
+ .get_or_default(PopulatesMetaFields)
+ .into();
+ if populates_meta_fields {
+ return Err(CoreError::Unsupported(
+ "append_only currently supports only tables with
hoodie.populate.meta.fields=false"
+ .to_string(),
+ ));
+ }
+ Ok(())
+ }
+
+ /// The metadata table is not updated, so its file listings would miss the
+ /// new base file.
+ async fn ensure_no_metadata_table(&self) -> Result<()> {
+ let partitions: Vec<String> = self
+ .table
+ .hudi_configs
+ .get_or_default(MetadataTablePartitions)
+ .into();
+ let has_metadata_dir = self
+ .table
+ .file_system_view
+ .storage
+ .list_dirs(Some(HUDI_METADATA_DIR))
+ .await?
+ .iter()
+ .any(|dir| dir == "metadata");
+ if partitions.iter().any(|p| !p.is_empty()) || has_metadata_dir {
+ return Err(CoreError::Unsupported(
+ "append_only currently supports only tables without a metadata
table".to_string(),
+ ));
+ }
+ Ok(())
+ }
+
+ fn is_layout_two(&self) -> bool {
+ let table_version: isize =
self.table.hudi_configs.get_or_default(TableVersion).into();
+ let layout_version: isize = self
+ .table
+ .hudi_configs
+ .get_or_default(TimelineLayoutVersion)
+ .into();
+ table_version >= 8 && layout_version == 2
+ }
+
+ fn timeline_dir(&self) -> String {
+ if self.is_layout_two() {
+ let timeline_path: String =
self.table.hudi_configs.get_or_default(TimelinePath).into();
+ format!("{HUDI_METADATA_DIR}/{timeline_path}")
+ } else {
+ HUDI_METADATA_DIR.to_string()
+ }
+ }
+
+ fn current_timestamp() -> String {
+ let now = Utc::now().timestamp_millis();
+
+ loop {
+ let last = LAST_EPOCH_MILLIS.load(Ordering::Relaxed);
+ let next = if now <= last { last + 1 } else { now };
+
+ if LAST_EPOCH_MILLIS
+ .compare_exchange(last, next, Ordering::SeqCst,
Ordering::Relaxed)
+ .is_ok()
+ {
+ return format!("{next:017}");
+ }
+ }
+ }
+
+ fn write_parquet_bytes(&self, batch: &RecordBatch) -> Result<Vec<u8>> {
+ let cursor = Cursor::new(Vec::new());
+ let mut writer = ArrowWriter::try_new(cursor, batch.schema(), None)?;
+ writer.write(batch)?;
+ let cursor = writer.into_inner()?;
+ Ok(cursor.into_inner())
+ }
+
+ async fn put_file(&self, relative_path: &str, bytes: Vec<u8>) ->
Result<()> {
+ let storage = self.table.file_system_view.storage.clone();
+ let object_url = join_url_segments(&storage.base_url,
&[relative_path])?;
+ let object_path =
Path::from_url_path(object_url.path()).map_err(StorageError::from)?;
+ storage
+ .object_store
+ .put(&object_path, PutPayload::from(Bytes::from(bytes)))
+ .await
+ .map_err(StorageError::from)?;
+ Ok(())
+ }
+
+ async fn delete_file(&self, relative_path: &str) -> Result<()> {
+ let storage = self.table.file_system_view.storage.clone();
+ let object_url = join_url_segments(&storage.base_url,
&[relative_path])?;
+ let object_path =
Path::from_url_path(object_url.path()).map_err(StorageError::from)?;
+ storage
+ .object_store
+ .delete(&object_path)
+ .await
+ .map_err(StorageError::from)?;
+ Ok(())
+ }
+
+ fn build_commit_metadata(
+ &self,
+ file_id: &str,
+ base_file_path: &str,
+ num_rows: i64,
+ file_size: i64,
+ ) -> HoodieCommitMetadata {
+ let write_stat = HoodieWriteStat {
+ file_id: Some(file_id.to_string()),
+ path: Some(base_file_path.to_string()),
+ base_file: None,
+ log_files: None,
+ prev_commit: None,
+ num_writes: Some(num_rows),
+ num_deletes: Some(0),
+ num_update_writes: Some(0),
+ num_inserts: Some(num_rows),
+ total_write_bytes: Some(file_size),
+ total_write_errors: Some(0),
+ partition_path: Some(String::new()),
+ total_log_records: None,
+ total_log_files: None,
+ total_updated_records_compacted: None,
+ total_log_blocks: None,
+ total_corrupt_log_block: None,
+ total_rollback_blocks: None,
+ file_size_in_bytes: Some(file_size),
+ log_version: None,
+ log_offset: None,
+ prev_base_file: None,
+ min_event_time: None,
+ max_event_time: None,
+ total_log_files_compacted: None,
+ total_log_read_time_ms: None,
+ total_log_size_compacted: None,
+ temp_path: None,
+ num_updates: Some(0),
+ };
+
+ HoodieCommitMetadata {
+ version: Some(1),
+ operation_type: Some("BULK_INSERT".to_string()),
+ partition_to_write_stats: Some(HashMap::from([(String::new(),
vec![write_stat])])),
+ partition_to_replace_file_ids: None,
+ compacted: Some(false),
+ extra_metadata: None,
+ }
+ }
+
+ async fn write_commit_file(
+ &self,
+ commit_relative_path: &str,
+ metadata: &HoodieCommitMetadata,
+ ) -> Result<()> {
+ let bytes = if self.is_layout_two() {
+ let schema = HoodieCommitMetadata::get_schema();
+ let mut writer = AvroWriter::new(&schema, Vec::new());
+ writer.append_ser(metadata)?;
+ writer.flush()?;
+ writer.into_inner()?
+ } else {
+ serde_json::to_vec(metadata).map_err(|e| {
+ CoreError::CommitMetadata(format!("Failed to serialize commit
metadata: {e}"))
+ })?
+ };
+
+ self.put_file(commit_relative_path, bytes).await
+ }
+}
diff --git a/crates/core/tests/table_bulk_insert_tests.rs
b/crates/core/tests/table_bulk_insert_tests.rs
new file mode 100644
index 00000000..8908ba2b
--- /dev/null
+++ b/crates/core/tests/table_bulk_insert_tests.rs
@@ -0,0 +1,386 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::fs;
+use std::sync::Arc;
+
+use arrow::array::{Int32Array, StringArray};
+use arrow::compute::concat_batches;
+use arrow::record_batch::RecordBatch;
+use arrow_schema::{DataType, Field, Schema};
+use hudi_core::error::CoreError;
+use hudi_core::error::Result;
+use hudi_core::table::{ReadOptions, Table};
+use tempfile::tempdir;
+
+fn write_table_properties(base_dir: &std::path::Path) -> Result<()> {
+ let hoodie_dir = base_dir.join(".hoodie");
+ fs::create_dir_all(&hoodie_dir)?;
+
+ let properties = [
+ "hoodie.table.name=test_table",
+ "hoodie.table.type=COPY_ON_WRITE",
+ "hoodie.table.version=6",
+ "hoodie.timeline.layout.version=1",
+ "hoodie.populate.meta.fields=false",
+ "hoodie.base.file.format=parquet",
+ ]
+ .join("\n");
+
+ fs::write(hoodie_dir.join("hoodie.properties"), properties)?;
+ Ok(())
+}
+
+fn write_partitioned_table_properties(base_dir: &std::path::Path) ->
Result<()> {
+ let hoodie_dir = base_dir.join(".hoodie");
+ fs::create_dir_all(&hoodie_dir)?;
+
+ let properties = [
+ "hoodie.table.name=test_table",
+ "hoodie.table.type=COPY_ON_WRITE",
+ "hoodie.table.version=6",
+ "hoodie.timeline.layout.version=1",
+ "hoodie.populate.meta.fields=false",
+ "hoodie.base.file.format=parquet",
+ "hoodie.table.partition.fields=region",
+ ]
+ .join("\n");
+
+ fs::write(hoodie_dir.join("hoodie.properties"), properties)?;
+ Ok(())
+}
+
+fn write_layout_v2_table_properties(base_dir: &std::path::Path) -> Result<()> {
+ let hoodie_dir = base_dir.join(".hoodie");
+ fs::create_dir_all(&hoodie_dir)?;
+
+ let properties = [
+ "hoodie.table.name=test_table",
+ "hoodie.table.type=COPY_ON_WRITE",
+ "hoodie.table.version=8",
+ "hoodie.timeline.layout.version=2",
+ "hoodie.populate.meta.fields=false",
+ "hoodie.base.file.format=parquet",
+ ]
+ .join("\n");
+
+ fs::write(hoodie_dir.join("hoodie.properties"), properties)?;
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_writes_records_and_commit() -> Result<()> {
+ let temp_dir = tempdir()?;
+ write_table_properties(temp_dir.path())?;
+
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ let schema = Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("name", DataType::Utf8, false),
+ ]);
+
+ let batch = RecordBatch::try_new(
+ schema.into(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["alice", "bob", "carol"])),
+ ],
+ )?;
+
+ let result = table.append_only([batch]).await?;
+ assert_eq!(result.num_rows, 3);
+ assert_eq!(
+ result.commit_relative_path,
+ format!(".hoodie/{}.commit", result.instant)
+ );
+
+ let reopened = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+ let records = reopened.read(&ReadOptions::new()).await?;
+ assert_eq!(records.len(), 1);
+
+ let read_batch = concat_batches(&records[0].schema(), &records)?;
+ assert_eq!(read_batch.num_rows(), 3);
+
+ let ids = read_batch
+ .column_by_name("id")
+ .and_then(|array| array.as_any().downcast_ref::<Int32Array>())
+ .expect("expected id column");
+ let names = read_batch
+ .column_by_name("name")
+ .and_then(|array| array.as_any().downcast_ref::<StringArray>())
+ .expect("expected name column");
+
+ assert_eq!(ids.values(), &[1, 2, 3]);
+ assert_eq!(names.value(0), "alice");
+ assert_eq!(names.value(1), "bob");
+ assert_eq!(names.value(2), "carol");
+
+ let commit_file = temp_dir.path().join(&result.commit_relative_path);
+ assert!(commit_file.exists(), "commit file should exist");
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_rejects_partitioned_tables() -> Result<()> {
+ let temp_dir = tempdir()?;
+ write_partitioned_table_properties(temp_dir.path())?;
+
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
+ let batch = RecordBatch::try_new(schema.into(),
vec![Arc::new(Int32Array::from(vec![1]))])?;
+
+ let error = table.append_only([batch]).await.unwrap_err();
+ assert!(
+ matches!(error, CoreError::Unsupported(message) if
message.contains("unpartitioned append-only tables"))
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_writes_layout_v2_commit() -> Result<()> {
+ let temp_dir = tempdir()?;
+ write_layout_v2_table_properties(temp_dir.path())?;
+
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ let schema = Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("name", DataType::Utf8, false),
+ ]);
+
+ let batch = RecordBatch::try_new(
+ schema.into(),
+ vec![
+ Arc::new(Int32Array::from(vec![7])),
+ Arc::new(StringArray::from(vec!["delta"])),
+ ],
+ )?;
+
+ let result = table.append_only([batch]).await?;
+ assert!(result.commit_relative_path.starts_with(".hoodie/timeline/"));
+ assert!(result.commit_relative_path.ends_with(".commit"));
+
+ let commit_file = temp_dir.path().join(&result.commit_relative_path);
+ assert!(commit_file.exists(), "layout v2 commit file should exist");
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_writes_multiple_batches_as_one_file() -> Result<()> {
+ let temp_dir = tempdir()?;
+ write_table_properties(temp_dir.path())?;
+
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("name", DataType::Utf8, false),
+ ]));
+ let first = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(StringArray::from(vec!["alice", "bob"])),
+ ],
+ )?;
+ let second = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![3])),
+ Arc::new(StringArray::from(vec!["carol"])),
+ ],
+ )?;
+
+ let result = table.append_only([first, second]).await?;
+ assert_eq!(result.num_rows, 3);
+
+ let reopened = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+ let records = reopened.read(&ReadOptions::new()).await?;
+ let read_batch = concat_batches(&records[0].schema(), &records)?;
+ let ids = read_batch
+ .column_by_name("id")
+ .and_then(|array| array.as_any().downcast_ref::<Int32Array>())
+ .expect("expected id column");
+ assert_eq!(ids.values(), &[1, 2, 3]);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_rejects_empty_and_mismatched_batches() -> Result<()> {
+ let temp_dir = tempdir()?;
+ write_table_properties(temp_dir.path())?;
+
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ let error = table
+ .append_only(Vec::<RecordBatch>::new())
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, CoreError::Write(message) if message.contains("at
least one RecordBatch"))
+ );
+
+ let ints = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])),
+ vec![Arc::new(Int32Array::from(vec![1]))],
+ )?;
+ let strings = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)])),
+ vec![Arc::new(StringArray::from(vec!["1"]))],
+ )?;
+ let error = table.append_only([ints, strings]).await.unwrap_err();
+ assert!(
+ matches!(error, CoreError::Write(message) if message.contains("share
the same schema"))
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_rejects_tables_with_meta_fields_or_metadata_table() ->
Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32,
false)]));
+ let batch = RecordBatch::try_new(schema,
vec![Arc::new(Int32Array::from(vec![1]))])?;
+
+ let meta_fields_dir = tempdir()?;
+ write_table_properties(meta_fields_dir.path())?;
+ let props = meta_fields_dir.path().join(".hoodie/hoodie.properties");
+ let rewritten = fs::read_to_string(&props)?.replace(
+ "hoodie.populate.meta.fields=false",
+ "hoodie.populate.meta.fields=true",
+ );
+ fs::write(&props, rewritten)?;
+ let mut table = Table::new(meta_fields_dir.path().to_str().expect("valid
temp path")).await?;
+ let error = table.append_only([batch.clone()]).await.unwrap_err();
+ assert!(
+ matches!(error, CoreError::Unsupported(message) if
message.contains("hoodie.populate.meta.fields=false"))
+ );
+
+ let mdt_dir = tempdir()?;
+ write_table_properties(mdt_dir.path())?;
+ fs::create_dir_all(mdt_dir.path().join(".hoodie/metadata/.hoodie"))?;
+ fs::write(
+ mdt_dir
+ .path()
+ .join(".hoodie/metadata/.hoodie/hoodie.properties"),
+ "hoodie.table.name=test_table_metadata\n",
+ )?;
+ let mut table = Table::new(mdt_dir.path().to_str().expect("valid temp
path")).await?;
+ let error = table.append_only([batch]).await.unwrap_err();
+ assert!(
+ matches!(error, CoreError::Unsupported(message) if
message.contains("without a metadata table"))
+ );
+ assert!(
+ fs::read_dir(mdt_dir.path().join(".hoodie"))?
+ .filter_map(|entry| entry.ok())
+ .all(|entry|
!entry.file_name().to_string_lossy().ends_with(".commit")),
+ "a rejected append must not write a commit"
+ );
+
+ Ok(())
+}
+
+fn id_name_batch(ids: Vec<i32>, names: Vec<&str>) -> RecordBatch {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("name", DataType::Utf8, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(StringArray::from(names)),
+ ],
+ )
+ .expect("valid batch")
+}
+
+#[tokio::test]
+async fn append_only_rejects_zero_rows_and_table_schema_mismatch() ->
Result<()> {
+ let temp_dir = tempdir()?;
+ write_table_properties(temp_dir.path())?;
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ let error = table
+ .append_only([id_name_batch(vec![], vec![])])
+ .await
+ .unwrap_err();
+ assert!(matches!(error, CoreError::Write(message) if message.contains("at
least one row")));
+
+ table
+ .append_only([id_name_batch(vec![1], vec!["alice"])])
+ .await?;
+ let ids_only = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])),
+ vec![Arc::new(Int32Array::from(vec![2]))],
+ )?;
+ let error = table.append_only([ids_only]).await.unwrap_err();
+ assert!(
+ matches!(error, CoreError::Write(message) if message.contains("does
not match table schema"))
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_handle_reads_its_own_writes() -> Result<()> {
+ let temp_dir = tempdir()?;
+ write_table_properties(temp_dir.path())?;
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ table
+ .append_only([id_name_batch(vec![1, 2], vec!["alice", "bob"])])
+ .await?;
+ table
+ .append_only([id_name_batch(vec![3], vec!["carol"])])
+ .await?;
+
+ let rows: usize = table
+ .read(&ReadOptions::new())
+ .await?
+ .iter()
+ .map(RecordBatch::num_rows)
+ .sum();
+ assert_eq!(rows, 3);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn append_only_accepts_cleared_metadata_partitions_property() ->
Result<()> {
+ let temp_dir = tempdir()?;
+ write_table_properties(temp_dir.path())?;
+ let props = temp_dir.path().join(".hoodie/hoodie.properties");
+ let mut content = fs::read_to_string(&props)?;
+ content.push_str("\nhoodie.table.metadata.partitions=\n");
+ fs::write(&props, content)?;
+ let mut table = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
+
+ let result = table
+ .append_only([id_name_batch(vec![1], vec!["alice"])])
+ .await?;
+ assert_eq!(result.num_rows, 1);
+
+ Ok(())
+}
