This is an automated email from the ASF dual-hosted git repository.
CTTY pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 94ceaf912 feat(writer): Support ObjectStorageLocationGenerator (#2871)
94ceaf912 is described below
commit 94ceaf9127fbb11dc85eb661c061b59b5f31c7ae
Author: hsiang-c <[email protected]>
AuthorDate: Tue Aug 4 14:54:20 2026 -0700
feat(writer): Support ObjectStorageLocationGenerator (#2871)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #2870
## What changes are included in this PR?
<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->
- Introduce `ObjectStorageLocationGenerator`, another implementation to
`LocationGenerator` that generates a deterministic hash for each stored
file and appends the hash directly after the `write.data.path`. The
equivalent class in Iceberg Java is [ObjectStoreLocationProvider
](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/LocationProviders.java#L128)
- I didn't make `ObjectStorageLocationGenerator` configurable yet. Plan
to do it in the follow-up PR.
## Are these changes tested?
<!--
Specify what test covers (unit test, integration test, etc.).
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
Unit tests
---------
Co-authored-by: Daniel Carl Jones <[email protected]>
---
crates/iceberg/public-api.txt | 21 ++
crates/iceberg/src/spec/table_properties.rs | 116 +++++--
crates/iceberg/src/util/location.rs | 44 +++
crates/iceberg/src/util/mod.rs | 2 +
.../src/writer/file_writer/location_generator.rs | 384 ++++++++++++++++++---
5 files changed, 491 insertions(+), 76 deletions(-)
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index d88ffa1b9..1f2829878 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -2764,9 +2764,13 @@ pub iceberg::spec::TableProperties::max_ref_age_ms: i64
pub iceberg::spec::TableProperties::max_snapshot_age_ms: i64
pub iceberg::spec::TableProperties::metadata_compression_codec:
iceberg::compression::CompressionCodec
pub iceberg::spec::TableProperties::min_snapshots_to_keep: usize
+pub iceberg::spec::TableProperties::write_data_location:
core::option::Option<alloc::string::String>
pub iceberg::spec::TableProperties::write_datafusion_fanout_enabled: bool
+pub iceberg::spec::TableProperties::write_folder_storage_location:
core::option::Option<alloc::string::String>
pub iceberg::spec::TableProperties::write_format_default: alloc::string::String
pub iceberg::spec::TableProperties::write_metadata_path:
core::option::Option<alloc::string::String>
+pub iceberg::spec::TableProperties::write_object_storage_location:
core::option::Option<alloc::string::String>
+pub iceberg::spec::TableProperties::write_object_storage_partitioned_paths:
bool
pub iceberg::spec::TableProperties::write_target_file_size_bytes: usize
impl iceberg::spec::TableProperties
pub const iceberg::spec::TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS:
&str
@@ -2814,7 +2818,12 @@ pub const
iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL: &str
pub const
iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32
pub const iceberg::spec::TableProperties::PROPERTY_SNAPSHOT_COUNT: &str
pub const iceberg::spec::TableProperties::PROPERTY_UUID: &str
+pub const iceberg::spec::TableProperties::PROPERTY_WRITE_DATA_LOCATION: &str
+pub const
iceberg::spec::TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION: &str
pub const iceberg::spec::TableProperties::PROPERTY_WRITE_METADATA_PATH: &str
+pub const
iceberg::spec::TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_LOCATION: &str
+pub const
iceberg::spec::TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS:
&str
+pub const
iceberg::spec::TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT:
bool
pub const
iceberg::spec::TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT: &str
pub const
iceberg::spec::TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT:
u64
pub const
iceberg::spec::TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES: &str
@@ -3181,6 +3190,7 @@ pub fn
iceberg::transform::TransformFunction::transform_literal_result(&self, in
pub fn iceberg::transform::create_transform_function(transform:
&iceberg::spec::Transform) ->
iceberg::Result<iceberg::transform::BoxedTransformFunction>
pub type iceberg::transform::BoxedTransformFunction = alloc::boxed::Box<dyn
iceberg::transform::TransformFunction>
pub mod iceberg::util
+pub mod iceberg::util::location
pub mod iceberg::util::snapshot
pub fn iceberg::util::snapshot::ancestors_between(table_metadata:
&iceberg::spec::TableMetadataRef, latest_snapshot_id: i64, oldest_snapshot_id:
core::option::Option<i64>) -> impl core::iter::traits::iterator::Iterator<Item
= iceberg::spec::SnapshotRef> + core::marker::Send
pub fn iceberg::util::snapshot::ancestors_of(table_metadata:
&iceberg::spec::TableMetadataRef, snapshot_id: i64) -> impl
core::iter::traits::iterator::Iterator<Item = iceberg::spec::SnapshotRef> +
core::marker::Send
@@ -3247,6 +3257,15 @@ impl core::fmt::Debug for
iceberg::writer::file_writer::location_generator::Defa
pub fn
iceberg::writer::file_writer::location_generator::DefaultLocationGenerator::fmt(&self,
f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
impl iceberg::writer::file_writer::location_generator::LocationGenerator for
iceberg::writer::file_writer::location_generator::DefaultLocationGenerator
pub fn
iceberg::writer::file_writer::location_generator::DefaultLocationGenerator::generate_location(&self,
partition_key: core::option::Option<&iceberg::spec::PartitionKey>, file_name:
&str) -> alloc::string::String
+pub struct
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator
+impl
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator
+pub fn
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator::new(table_metadata:
&iceberg::spec::TableMetadata) -> iceberg::Result<Self>
+impl core::clone::Clone for
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator
+pub fn
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator::clone(&self)
->
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator
+impl core::fmt::Debug for
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator
+pub fn
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator::fmt(&self,
f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl iceberg::writer::file_writer::location_generator::LocationGenerator for
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator
+pub fn
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator::generate_location(&self,
partition_key: core::option::Option<&iceberg::spec::PartitionKey>, file_name:
&str) -> alloc::string::String
pub trait iceberg::writer::file_writer::location_generator::FileNameGenerator:
core::clone::Clone + core::marker::Send + core::marker::Sync + 'static
pub fn
iceberg::writer::file_writer::location_generator::FileNameGenerator::generate_file_name(&self)
-> alloc::string::String
impl iceberg::writer::file_writer::location_generator::FileNameGenerator for
iceberg::writer::file_writer::location_generator::DefaultFileNameGenerator
@@ -3255,6 +3274,8 @@ pub trait
iceberg::writer::file_writer::location_generator::LocationGenerator: c
pub fn
iceberg::writer::file_writer::location_generator::LocationGenerator::generate_location(&self,
partition_key: core::option::Option<&iceberg::spec::PartitionKey>, file_name:
&str) -> alloc::string::String
impl iceberg::writer::file_writer::location_generator::LocationGenerator for
iceberg::writer::file_writer::location_generator::DefaultLocationGenerator
pub fn
iceberg::writer::file_writer::location_generator::DefaultLocationGenerator::generate_location(&self,
partition_key: core::option::Option<&iceberg::spec::PartitionKey>, file_name:
&str) -> alloc::string::String
+impl iceberg::writer::file_writer::location_generator::LocationGenerator for
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator
+pub fn
iceberg::writer::file_writer::location_generator::ObjectStorageLocationGenerator::generate_location(&self,
partition_key: core::option::Option<&iceberg::spec::PartitionKey>, file_name:
&str) -> alloc::string::String
pub mod iceberg::writer::file_writer::rolling_writer
pub struct iceberg::writer::file_writer::rolling_writer::RollingFileWriter<B:
iceberg::writer::file_writer::FileWriterBuilder, L:
iceberg::writer::file_writer::location_generator::LocationGenerator, F:
iceberg::writer::file_writer::location_generator::FileNameGenerator>
impl<B, L, F>
iceberg::writer::file_writer::rolling_writer::RollingFileWriter<B, L, F> where
B: iceberg::writer::file_writer::FileWriterBuilder, L:
iceberg::writer::file_writer::location_generator::LocationGenerator, F:
iceberg::writer::file_writer::location_generator::FileNameGenerator
diff --git a/crates/iceberg/src/spec/table_properties.rs
b/crates/iceberg/src/spec/table_properties.rs
index 379feee5c..a35f1427a 100644
--- a/crates/iceberg/src/spec/table_properties.rs
+++ b/crates/iceberg/src/spec/table_properties.rs
@@ -21,6 +21,7 @@ use std::str::FromStr;
use crate::compression::CompressionCodec;
use crate::error::{Error, ErrorKind, Result};
+use crate::util::location::strip_trailing_slash;
fn parse_property<T: FromStr>(
properties: &HashMap<String, String>,
@@ -40,18 +41,6 @@ where
})
}
-/// Strips trailing slashes from a location, preserving a bare URI scheme root
-fn strip_trailing_slash(path: &str) -> &str {
- let mut path = path;
- while !path.ends_with("://") {
- let Some(stripped) = path.strip_suffix('/') else {
- break;
- };
- path = stripped;
- }
- path
-}
-
fn parse_location_property(
properties: &HashMap<String, String>,
key: &str,
@@ -128,6 +117,24 @@ pub(crate) fn parse_metadata_file_compression(
}
}
+/// Parse boolean property case insensitively
+/// Rust standard library only accepts "true" and "false", see
https://doc.rust-lang.org/std/primitive.bool.html#method.from_str
+/// Users might accidentally trigger fallback with valid configuration values
such as "False" or "True"
+fn parse_property_bool(
+ properties: &HashMap<String, String>,
+ key: &str,
+ default: bool,
+) -> Result<bool> {
+ properties.get(key).map_or(Ok(default), |value| {
+ value.to_lowercase().parse::<bool>().map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid value for {key}: {e}"),
+ )
+ })
+ })
+}
+
/// TableProperties that contains the properties of a table.
#[derive(Debug)]
pub struct TableProperties {
@@ -173,6 +180,20 @@ pub struct TableProperties {
pub encryption_key_id: Option<String>,
/// The encryption data encryption key length in bytes.
pub encryption_data_key_length: usize,
+ /// Base directory for data files
+ pub write_data_location: Option<String>,
+ /// Deprecated table property for data file write location.
+ ///
+ /// Property will be removed at a later date.
+ /// Superseded by [write_data_location].
+ pub write_folder_storage_location: Option<String>,
+ /// Deprecated table property for data file write location for object
storage location generator.
+ ///
+ /// Property will be removed at a later date.
+ /// Superseded by [write_data_location].
+ pub write_object_storage_location: Option<String>,
+ /// Whether partition values are included in object storage paths.
+ pub write_object_storage_partitioned_paths: bool,
}
impl TableProperties {
@@ -325,6 +346,17 @@ impl TableProperties {
pub const PROPERTY_ENCRYPTION_DATA_KEY_LENGTH: &str =
"encryption.data-key-length";
/// Default value for the encryption DEK length (16 bytes = AES-128).
pub const PROPERTY_ENCRYPTION_DATA_KEY_LENGTH_DEFAULT: usize = 16;
+ /// Property key for the base directory for data files
+ pub const PROPERTY_WRITE_DATA_LOCATION: &str = "write.data.path";
+ /// Property key for deprecated [write_folder_storage_location]
+ pub const PROPERTY_WRITE_FOLDER_STORAGE_LOCATION: &str =
"write.folder-storage.path";
+ /// Property key for deprecated object storage path, kept as a fallback
for compatibility.
+ pub const PROPERTY_WRITE_OBJECT_STORAGE_LOCATION: &str =
"write.object-storage.path";
+ /// Property key for controlling whether partition values are included in
object storage paths.
+ pub const PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS: &str =
+ "write.object-storage.partitioned-paths";
+ /// Default value for [PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS]
+ pub const PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT: bool =
true;
}
impl TryFrom<&HashMap<String, String>> for TableProperties {
@@ -368,12 +400,12 @@ impl TryFrom<&HashMap<String, String>> for
TableProperties {
TableProperties::PROPERTY_WRITE_METADATA_PATH,
)?,
metadata_compression_codec:
parse_metadata_file_compression(props)?,
- write_datafusion_fanout_enabled: parse_property(
+ write_datafusion_fanout_enabled: parse_property_bool(
props,
TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED,
TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED_DEFAULT,
)?,
- gc_enabled: parse_property(
+ gc_enabled: parse_property_bool(
props,
TableProperties::PROPERTY_GC_ENABLED,
TableProperties::PROPERTY_GC_ENABLED_DEFAULT,
@@ -393,7 +425,7 @@ impl TryFrom<&HashMap<String, String>> for TableProperties {
TableProperties::PROPERTY_MAX_REF_AGE_MS,
TableProperties::PROPERTY_MAX_REF_AGE_MS_DEFAULT,
)?,
- cdc_enabled: parse_property(
+ cdc_enabled: parse_property_bool(
props,
TableProperties::PROPERTY_PARQUET_CDC_ENABLED,
TableProperties::PROPERTY_PARQUET_CDC_ENABLED_DEFAULT,
@@ -421,6 +453,20 @@ impl TryFrom<&HashMap<String, String>> for TableProperties
{
TableProperties::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH,
TableProperties::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH_DEFAULT,
)?,
+ write_data_location: props
+ .get(TableProperties::PROPERTY_WRITE_DATA_LOCATION)
+ .cloned(),
+ write_folder_storage_location: props
+ .get(TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION)
+ .cloned(),
+ write_object_storage_location: props
+ .get(TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_LOCATION)
+ .cloned(),
+ write_object_storage_partitioned_paths: parse_property_bool(
+ props,
+
TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS,
+
TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT,
+ )?,
})
}
}
@@ -529,22 +575,6 @@ mod tests {
);
}
- #[test]
- fn test_strip_trailing_slash() {
- for (path, expected) in [
- ("s3://bucket/db/tbl", "s3://bucket/db/tbl"),
- ("s3://bucket/db/tbl/", "s3://bucket/db/tbl"),
- ("s3://bucket/db/tbl////", "s3://bucket/db/tbl"),
- ("blobstore://", "blobstore://"),
- ("blobstore:///", "blobstore://"),
- ("file:///", "file://"),
- ("////", ""),
- ("", ""),
- ] {
- assert_eq!(strip_trailing_slash(path), expected);
- }
- }
-
#[test]
fn test_table_properties_compression() {
let props = HashMap::from([(
@@ -947,4 +977,28 @@ mod tests {
let tp = TableProperties::try_from(&props).unwrap();
assert!(!tp.cdc_enabled);
}
+
+ #[test]
+ fn test_parse_boolean_property_case_insensitive() {
+ let false_variants = ["False", "FALSE"];
+ let true_variants = ["True", "TRUE"];
+
+ for f in false_variants {
+ let props = HashMap::from([(
+
TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS.to_string(),
+ f.to_string(),
+ )]);
+ let tp = TableProperties::try_from(&props).unwrap();
+ assert!(!tp.write_object_storage_partitioned_paths);
+ }
+
+ for t in true_variants {
+ let props = HashMap::from([(
+
TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS.to_string(),
+ t.to_string(),
+ )]);
+ let tp = TableProperties::try_from(&props).unwrap();
+ assert!(tp.write_object_storage_partitioned_paths);
+ }
+ }
}
diff --git a/crates/iceberg/src/util/location.rs
b/crates/iceberg/src/util/location.rs
new file mode 100644
index 000000000..00681a12a
--- /dev/null
+++ b/crates/iceberg/src/util/location.rs
@@ -0,0 +1,44 @@
+// 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.
+
+/// Strips trailing slashes from a location, preserving a bare URI scheme root
+pub(crate) fn strip_trailing_slash(path: &str) -> &str {
+ let mut path = path;
+ while !path.ends_with("://") {
+ let Some(stripped) = path.strip_suffix('/') else {
+ break;
+ };
+ path = stripped;
+ }
+ path
+}
+
+#[test]
+fn test_strip_trailing_slash() {
+ for (path, expected) in [
+ ("s3://bucket/db/tbl", "s3://bucket/db/tbl"),
+ ("s3://bucket/db/tbl/", "s3://bucket/db/tbl"),
+ ("s3://bucket/db/tbl////", "s3://bucket/db/tbl"),
+ ("blobstore://", "blobstore://"),
+ ("blobstore:///", "blobstore://"),
+ ("file:///", "file://"),
+ ("////", ""),
+ ("", ""),
+ ] {
+ assert_eq!(strip_trailing_slash(path), expected);
+ }
+}
diff --git a/crates/iceberg/src/util/mod.rs b/crates/iceberg/src/util/mod.rs
index 3cf2eef9b..94da0130c 100644
--- a/crates/iceberg/src/util/mod.rs
+++ b/crates/iceberg/src/util/mod.rs
@@ -17,6 +17,8 @@
use std::num::NonZeroUsize;
+/// Utilities for working with data, metadata, database, table ...etc.
locations
+pub mod location;
/// Utilities for working with snapshots.
pub mod snapshot;
diff --git a/crates/iceberg/src/writer/file_writer/location_generator.rs
b/crates/iceberg/src/writer/file_writer/location_generator.rs
index d37d00c93..caaeb7b68 100644
--- a/crates/iceberg/src/writer/file_writer/location_generator.rs
+++ b/crates/iceberg/src/writer/file_writer/location_generator.rs
@@ -21,7 +21,8 @@ use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use crate::Result;
-use crate::spec::{DataFileFormat, PartitionKey, TableMetadata};
+use crate::spec::{DataFileFormat, PartitionKey, TableMetadata,
TableProperties};
+use crate::util::location::strip_trailing_slash;
/// `LocationGenerator` used to generate the location of data file.
pub trait LocationGenerator: Clone + Send + Sync + 'static {
@@ -41,9 +42,14 @@ pub trait LocationGenerator: Clone + Send + Sync + 'static {
fn generate_location(&self, partition_key: Option<&PartitionKey>,
file_name: &str) -> String;
}
-const WRITE_DATA_LOCATION: &str = "write.data.path";
-const WRITE_FOLDER_STORAGE_LOCATION: &str = "write.folder-storage.path";
+/// Default directory for data files
const DEFAULT_DATA_DIR: &str = "/data";
+/// Number of trailing hash bits used to build the entropy directories.
+const HASH_BINARY_STRING_BITS: usize = 20;
+/// Length of each entropy directory.
+const ENTROPY_DIR_LENGTH: usize = 4;
+/// Number of entropy directories generated from the hash.
+const ENTROPY_DIR_DEPTH: usize = 3;
#[derive(Clone, Debug)]
/// `DefaultLocationGenerator` used to generate the data dir location of data
file.
@@ -54,17 +60,22 @@ pub struct DefaultLocationGenerator {
impl DefaultLocationGenerator {
/// Create a new `DefaultLocationGenerator`.
+ /// Resolve the base data location from table properties, falling back to
`{table_location}/data`.
+ ///
+ /// Precedence follows Java Iceberg: `write.data.path` first, then the
deprecated
+ /// `write.folder-storage.path` properties, and finally the default
+ /// `{table_location}/data`.
pub fn new(table_metadata: &TableMetadata) -> Result<Self> {
- let table_location = table_metadata.location();
- let prop = table_metadata.properties();
- let configured_data_location = prop
- .get(WRITE_DATA_LOCATION)
- .or(prop.get(WRITE_FOLDER_STORAGE_LOCATION));
- let data_location = if let Some(data_location) =
configured_data_location {
- data_location.clone()
- } else {
- format!("{table_location}{DEFAULT_DATA_DIR}")
- };
+ let table_location = strip_trailing_slash(table_metadata.location());
+ let prop = TableProperties::try_from(table_metadata.properties())?;
+ let data_location = strip_trailing_slash(
+ prop.write_data_location
+ .or(prop.write_folder_storage_location)
+ .unwrap_or(format!("{table_location}{DEFAULT_DATA_DIR}"))
+ .as_ref(),
+ )
+ .to_string();
+
Ok(Self { data_location })
}
@@ -93,6 +104,136 @@ impl LocationGenerator for DefaultLocationGenerator {
}
}
+/// `ObjectStorageLocationGenerator` injects hash entropy into generated file
locations so that
+/// files are spread across many object-store prefixes.
+///
+/// Object stores such as S3 shard request throughput by key prefix, so
writing every file under a
+/// common `.../data/` prefix creates a throughput hotspot. This generator
prepends a
+/// deterministic, hashed directory tree (derived from the file name) to each
location, mirroring
+/// Java Iceberg's `ObjectStoreLocationProvider`.
+///
+/// The behavior is controlled by these table properties:
+/// * `write.data.path` / `write.object-storage.path` /
`write.folder-storage.path` - the base data
+/// location (checked in that order), defaulting to `{table_location}/data`.
+/// * `write.object-storage.partitioned-paths` - whether partition values are
included in the path
+/// (defaults to `true`).
+#[derive(Clone, Debug)]
+pub struct ObjectStorageLocationGenerator {
+ storage_location: String,
+ /// Database/table context, only set when the storage location is outside
the table location.
+ context: Option<String>,
+ include_partition_paths: bool,
+}
+
+impl ObjectStorageLocationGenerator {
+ /// Create a new `ObjectStorageLocationGenerator` from table metadata.
+ /// Resolve the base data location from table properties, falling back to
`{table_location}/data`.
+ ///
+ /// Precedence follows Java Iceberg: `write.data.path` first, then the
deprecated
+ /// `write.object-storage.path` and `write.folder-storage.path`
properties, and finally the default
+ /// `{table_location}/data`.
+ pub fn new(table_metadata: &TableMetadata) -> Result<Self> {
+ let table_location = strip_trailing_slash(table_metadata.location());
+ let prop = TableProperties::try_from(table_metadata.properties())?;
+ let storage_location = strip_trailing_slash(
+ prop.write_data_location
+ .or(prop.write_object_storage_location)
+ .or(prop.write_folder_storage_location)
+ .unwrap_or(format!("{table_location}{DEFAULT_DATA_DIR}"))
+ .as_ref(),
+ )
+ .to_string();
+
+ // If the storage location is within the table prefix, files are
already scoped to this
+ // table so there is no need to add database/table context to avoid
collisions.
+ let context = if storage_location.starts_with(table_location) {
+ None
+ } else {
+ Some(path_context(table_location))
+ };
+
+ let include_partition_paths =
prop.write_object_storage_partitioned_paths;
+
+ Ok(Self {
+ storage_location,
+ context,
+ include_partition_paths,
+ })
+ }
+
+ /// Build the final location for a fully-formed data file name (which may
already include a
+ /// partition path).
+ fn new_data_location(&self, name: &str) -> String {
+ let hash = self.compute_hash(name);
+ if let Some(context) = &self.context {
+ format!("{}/{}/{}/{}", self.storage_location, hash, context, name)
+ } else if self.include_partition_paths {
+ format!("{}/{}/{}", self.storage_location, hash, name)
+ } else {
+ // When partition paths are excluded, join the entropy to the file
name with `-` so the
+ // file still lives directly under the storage location.
+ format!("{}/{}-{}", self.storage_location, hash, name)
+ }
+ }
+
+ /// Compute the entropy directory tree for the given name, e.g.
`0101/0110/1001/10110010`.
+ fn compute_hash(&self, name: &str) -> String {
+ let mut bytes = name.as_bytes();
+ let hash_code = murmur3::murmur3_32(&mut bytes, 0).unwrap();
+ let binary = format!("{:032b}", hash_code);
+ let hash = &binary[binary.len() - HASH_BINARY_STRING_BITS..];
+ dirs_from_hash(hash)
+ }
+}
+
+impl LocationGenerator for ObjectStorageLocationGenerator {
+ fn generate_location(&self, partition_key: Option<&PartitionKey>,
file_name: &str) -> String {
+ let name =
+ if self.include_partition_paths &&
!PartitionKey::is_effectively_none(partition_key) {
+ format!("{}/{}", partition_key.unwrap().to_path(), file_name)
+ } else {
+ file_name.to_string()
+ };
+ self.new_data_location(&name)
+ }
+}
+
+/// Derive the `{parent}/{name}` context from a table location, mirroring
Hadoop's
+/// `Path.getParent().getName()` / `Path.getName()`.
+fn path_context(table_location: &str) -> String {
+ let mut segments = table_location.rsplit('/').filter(|s| !s.is_empty());
+ let name = segments.next().unwrap_or("");
+ match segments.next() {
+ Some(parent) => format!("{parent}/{name}"),
+ None => name.to_string(),
+ }
+}
+
+/// Divide a binary hash string into directories for optimized listing/orphan
removal.
+///
+/// With `ENTROPY_DIR_DEPTH = 3` and `ENTROPY_DIR_LENGTH = 4`, the 20-bit hash
+/// `10011001100110011001` becomes `1001/1001/1001/10011001`.
+fn dirs_from_hash(hash: &str) -> String {
+ let mut result = String::new();
+
+ let mut i = 0;
+ while i < ENTROPY_DIR_DEPTH * ENTROPY_DIR_LENGTH {
+ if i > 0 {
+ result.push('/');
+ }
+ let end = (i + ENTROPY_DIR_LENGTH).min(hash.len());
+ result.push_str(&hash[i..end]);
+ i += ENTROPY_DIR_LENGTH;
+ }
+
+ if hash.len() > ENTROPY_DIR_DEPTH * ENTROPY_DIR_LENGTH {
+ result.push('/');
+ result.push_str(&hash[ENTROPY_DIR_DEPTH * ENTROPY_DIR_LENGTH..]);
+ }
+
+ result
+}
+
/// `FileNameGeneratorTrait` used to generate file name for data file. The
file name can be passed to `LocationGenerator` to generate the location of the
file.
pub trait FileNameGenerator: Clone + Send + Sync + 'static {
/// Generate a file name.
@@ -150,41 +291,15 @@ pub(crate) mod test {
use super::LocationGenerator;
use crate::spec::{
FormatVersion, Literal, NestedField, PartitionKey, PartitionSpec,
PrimitiveType, Schema,
- Struct, StructType, TableMetadata, Transform, Type,
+ Struct, StructType, TableMetadata, TableProperties, Transform, Type,
};
use crate::writer::file_writer::location_generator::{
- DefaultLocationGenerator, FileNameGenerator, WRITE_DATA_LOCATION,
- WRITE_FOLDER_STORAGE_LOCATION,
+ DefaultLocationGenerator, FileNameGenerator,
ObjectStorageLocationGenerator,
};
#[test]
fn test_default_location_generate() {
- let mut table_metadata = TableMetadata {
- format_version: FormatVersion::V2,
- table_uuid:
Uuid::parse_str("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94").unwrap(),
- location: "s3://data.db/table".to_string(),
- last_updated_ms: 1515100955770,
- last_column_id: 1,
- schemas: HashMap::new(),
- current_schema_id: 1,
- partition_specs: HashMap::new(),
- default_spec: PartitionSpec::unpartition_spec().into(),
- default_partition_type: StructType::new(vec![]),
- last_partition_id: 1000,
- default_sort_order_id: 0,
- sort_orders: HashMap::from_iter(vec![]),
- snapshots: HashMap::default(),
- current_snapshot_id: None,
- last_sequence_number: 1,
- properties: HashMap::new(),
- snapshot_log: Vec::new(),
- metadata_log: vec![],
- refs: HashMap::new(),
- statistics: HashMap::new(),
- partition_statistics: HashMap::new(),
- encryption_keys: HashMap::new(),
- next_row_id: 0,
- };
+ let mut table_metadata = table_metadata_with("s3://data.db/table",
HashMap::new());
let file_name_generator = super::DefaultFileNameGenerator::new(
"part".to_string(),
@@ -200,7 +315,7 @@ pub(crate) mod test {
// test custom data location
table_metadata.properties.insert(
- WRITE_FOLDER_STORAGE_LOCATION.to_string(),
+
TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION.to_string(),
"s3://data.db/table/data_1".to_string(),
);
let location_generator =
DefaultLocationGenerator::new(&table_metadata).unwrap();
@@ -212,7 +327,7 @@ pub(crate) mod test {
);
table_metadata.properties.insert(
- WRITE_DATA_LOCATION.to_string(),
+ TableProperties::PROPERTY_WRITE_DATA_LOCATION.to_string(),
"s3://data.db/table/data_2".to_string(),
);
let location_generator =
DefaultLocationGenerator::new(&table_metadata).unwrap();
@@ -224,7 +339,7 @@ pub(crate) mod test {
);
table_metadata.properties.insert(
- WRITE_DATA_LOCATION.to_string(),
+ TableProperties::PROPERTY_WRITE_DATA_LOCATION.to_string(),
// invalid table location
"s3://data.db/data_3".to_string(),
);
@@ -310,6 +425,185 @@ pub(crate) mod test {
);
}
+ // This test is ported from
https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/test/java/org/apache/iceberg/TestLocationProvider.java#L303
+ #[test]
+ fn test_object_storage_hash_injection() {
+ // Golden vectors ported from Java's
TestLocationProvider#testHashInjection, verifying that
+ // the murmur3 entropy directories match Java Iceberg exactly.
+ let table_metadata = table_metadata_with("s3://data.db/table",
HashMap::new());
+ let location_gen =
ObjectStorageLocationGenerator::new(&table_metadata).unwrap();
+
+ for (file_name, expected) in [
+ ("a", "s3://data.db/table/data/0101/0110/1001/10110010/a"),
+ ("b", "s3://data.db/table/data/1110/0111/1110/00000011/b"),
+ ("c", "s3://data.db/table/data/0010/1101/0110/01011111/c"),
+ ("d", "s3://data.db/table/data/1001/0001/0100/01110011/d"),
+ ] {
+ assert_eq!(location_gen.generate_location(None, file_name),
expected);
+ }
+ }
+
+ #[test]
+ fn test_object_storage_include_partition_paths() {
+ // With partitioned paths enabled (the default), the partition path is
part of the hashed
+ // name and appears after the entropy directories.
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ NestedField::required(2, "a",
Type::Primitive(PrimitiveType::String)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+ let partition_spec = PartitionSpec::builder(schema.clone())
+ .add_partition_field("id", "id", Transform::Identity)
+ .unwrap()
+ .add_partition_field("a", "a_trunc_3", Transform::Truncate(3))
+ .unwrap()
+ .build()
+ .unwrap();
+ let partition_data =
+ Struct::from_iter([Some(Literal::int(0)),
Some(Literal::string("apa"))]);
+ let partition_key = PartitionKey::new(partition_spec, schema,
partition_data);
+
+ let table_metadata = table_metadata_with("s3://data.db/table",
HashMap::new());
+ let location_gen =
ObjectStorageLocationGenerator::new(&table_metadata).unwrap();
+ let location = location_gen.generate_location(Some(&partition_key),
"test.parquet");
+
+ assert!(
+ location.starts_with("s3://data.db/table/data/"),
+ "unexpected location: {location}"
+ );
+ // The partition path is included
+ assert!(
+ location.ends_with("/id=0/a_trunc_3=apa/test.parquet"),
+ "unexpected location: {location}"
+ );
+ }
+
+ // This test is ported from
https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/test/java/org/apache/iceberg/TestLocationProvider.java#L267
+ #[test]
+ fn test_object_storage_include_partition_paths_with_special_characters() {
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "data#1",
Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+ let partition_spec = PartitionSpec::builder(schema.clone())
+ .add_partition_field("data#1", "data#1", Transform::Identity)
+ .unwrap()
+ .build()
+ .unwrap();
+ let partition_data =
Struct::from_iter([Some(Literal::string("val#1"))]);
+ let partition_key = PartitionKey::new(partition_spec, schema,
partition_data);
+
+ let table_metadata = table_metadata_with(
+ "s3://data.db/table",
+ HashMap::from([(
+
TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS.to_string(),
+ "true".to_string(),
+ )]),
+ );
+ let location_gen =
ObjectStorageLocationGenerator::new(&table_metadata).unwrap();
+ let location = location_gen.generate_location(Some(&partition_key),
"test.parquet");
+
+ assert_eq!(
+ location,
+
"s3://data.db/table/data/0000/1011/0110/00001000/data%231=val%231/test.parquet"
+ );
+ }
+
+ // This test is ported from
https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/test/java/org/apache/iceberg/TestLocationProvider.java#L285
+ #[test]
+ fn test_object_storage_exclude_partition_paths() {
+ // Golden vector ported from Java's
TestLocationProvider#testExcludePartitionInPath. With
+ // partitioned paths disabled, the partition value is dropped and the
last entropy dir is
+ // joined to the file name with `-`.
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+ let partition_spec = PartitionSpec::builder(schema.clone())
+ .add_partition_field("id", "id", Transform::Identity)
+ .unwrap()
+ .build()
+ .unwrap();
+ let partition_data = Struct::from_iter([Some(Literal::int(0))]);
+ let partition_key = PartitionKey::new(partition_spec, schema,
partition_data);
+
+ let table_metadata = table_metadata_with(
+ "s3://data.db/table",
+ HashMap::from([(
+
TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS.to_string(),
+ "false".to_string(),
+ )]),
+ );
+ let location_gen =
ObjectStorageLocationGenerator::new(&table_metadata).unwrap();
+ let location = location_gen.generate_location(Some(&partition_key),
"test.parquet");
+
+ assert_eq!(
+ location,
+ "s3://data.db/table/data/0110/1010/0011/11101000-test.parquet"
+ );
+ }
+
+ #[test]
+ fn test_object_storage_context_when_data_location_outside_table() {
+ // When the data location is outside the table location, the
database/table context is
+ // injected after the entropy directories to avoid cross-table
collisions.
+ let table_metadata = table_metadata_with(
+ "s3://data.db/table",
+ HashMap::from([(
+ TableProperties::PROPERTY_WRITE_DATA_LOCATION.to_string(),
+ "s3://custom-bucket/objects".to_string(),
+ )]),
+ );
+ let location_gen =
ObjectStorageLocationGenerator::new(&table_metadata).unwrap();
+ let location = location_gen.generate_location(None, "a");
+
+ // Entropy for "a" is 0101/0110/1001/10110010, then the
"data.db/table" context, then file.
+ assert_eq!(
+ location,
+
"s3://custom-bucket/objects/0101/0110/1001/10110010/data.db/table/a"
+ );
+ }
+
+ #[test]
+ fn test_object_storage_data_location_precedence() {
+ // write.data.path takes precedence over the deprecated folder-storage
fallback.
+ let table_metadata = table_metadata_with(
+ "s3://data.db/table",
+ HashMap::from([
+ (
+ TableProperties::PROPERTY_WRITE_DATA_LOCATION.to_string(),
+ "s3://data.db/table/data_primary".to_string(),
+ ),
+ (
+
TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION.to_string(),
+ "s3://data.db/table/data_legacy".to_string(),
+ ),
+ ]),
+ );
+ let location_gen =
ObjectStorageLocationGenerator::new(&table_metadata).unwrap();
+ let location = location_gen.generate_location(None, "a");
+
+ assert_eq!(
+ location,
+ "s3://data.db/table/data_primary/0101/0110/1001/10110010/a"
+ );
+ }
+
/// Build a minimal `TableMetadata` for location generator tests.
fn table_metadata_with(location: &str, properties: HashMap<String,
String>) -> TableMetadata {
TableMetadata {
