CTTY commented on code in PR #2871:
URL: https://github.com/apache/iceberg-rust/pull/2871#discussion_r3634230809
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -93,6 +106,153 @@ 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.
+ pub fn new(table_metadata: &TableMetadata) -> Result<Self> {
+ let table_location = strip_trailing_slash(table_metadata.location());
+ let prop = table_metadata.properties();
+ let storage_location =
+ strip_trailing_slash(&resolve_data_location(prop,
table_location)).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
+ .get(WRITE_OBJECT_STORAGE_PARTITIONED_PATHS)
+ .and_then(|value| value.parse::<bool>().ok())
+ .unwrap_or(WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT);
+
+ 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();
+
+ // Force the top bit so the binary string always has 32 characters
(Rust, like Java's
+ // Integer.toBinaryString, drops leading zeros otherwise), then keep
the trailing bits.
+ let binary = format!("{:032b}", hash_code | 0x8000_0000);
+ 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)
+ }
+}
+
+/// Strip a single trailing slash from a location, if present.
+fn strip_trailing_slash(location: &str) -> &str {
Review Comment:
in table_properties.rs, there is already one `strip_trailing_slash`, can we
refactor and use that? We can create a new `iceberg/src/util/location.rs` to
store such helpers
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -45,6 +45,19 @@ const WRITE_DATA_LOCATION: &str = "write.data.path";
const WRITE_FOLDER_STORAGE_LOCATION: &str = "write.folder-storage.path";
const DEFAULT_DATA_DIR: &str = "/data";
+/// Deprecated object storage path property, kept as a fallback for
compatibility.
+const WRITE_OBJECT_STORAGE_LOCATION: &str = "write.object-storage.path";
+/// Property controlling whether partition values are included in object
storage paths.
+const WRITE_OBJECT_STORAGE_PARTITIONED_PATHS: &str =
"write.object-storage.partitioned-paths";
+const WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT: bool = true;
+
+/// 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;
Review Comment:
We should move all of these to table_properties.rs
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -93,6 +106,153 @@ 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.
+ pub fn new(table_metadata: &TableMetadata) -> Result<Self> {
+ let table_location = strip_trailing_slash(table_metadata.location());
+ let prop = table_metadata.properties();
+ let storage_location =
+ strip_trailing_slash(&resolve_data_location(prop,
table_location)).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
+ .get(WRITE_OBJECT_STORAGE_PARTITIONED_PATHS)
+ .and_then(|value| value.parse::<bool>().ok())
+ .unwrap_or(WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT);
+
+ 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();
+
+ // Force the top bit so the binary string always has 32 characters
(Rust, like Java's
+ // Integer.toBinaryString, drops leading zeros otherwise), then keep
the trailing bits.
Review Comment:
I think `{:032b}` already forces the binary to be 32 characters with zero
paddings? In java we have to do `| 0x8000_0000` because the hash to binary
string conversion won't preserve the leading zeroes, but here looks like we
don't actually need to set the first bit to 1?
The logic is ok, but I think it's worth clarifying in the comment that we
are setting the first bit to 1 simply to imitate java's padding logic
--
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]
