This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new a7592ed0f fix(table)!: honor object storage partitioned-paths for
partitioned writes (#2368)
a7592ed0f is described below
commit a7592ed0f311a2b5248e0a14f1f40fcd3bb80ca2
Author: Ish <[email protected]>
AuthorDate: Wed Nov 4 16:50:13 2026 +0100
fix(table)!: honor object storage partitioned-paths for partitioned writes
(#2368)
---
table/locations.go | 31 ++++++-----
table/locations_test.go | 43 +++++++--------
table/rolling_data_writer.go | 56 +++----------------
table/rolling_data_writer_test.go | 111 ++++++++++++++++++++++++--------------
table/writer.go | 2 +-
website/src/releases.md | 2 +
6 files changed, 123 insertions(+), 122 deletions(-)
diff --git a/table/locations.go b/table/locations.go
index a051c291f..f782bf2f5 100644
--- a/table/locations.go
+++ b/table/locations.go
@@ -19,7 +19,6 @@ package table
import (
"fmt"
- "path"
"strings"
"github.com/apache/iceberg-go"
@@ -38,7 +37,9 @@ const (
)
type LocationProvider interface {
- NewDataLocation(dataFileName string) string
+ // NewDataLocation returns the location for a data file. partitionPath
is the
+ // file's partition path, such as "a=1/b=2", or empty for an
unpartitioned file.
+ NewDataLocation(dataFileName, partitionPath string) string
NewTableMetadataFileLocation(newVersion int) (string, error)
NewMetadataLocation(metadataFileName string) string
}
@@ -79,8 +80,12 @@ func joinLocationPath(location string, elem ...string)
string {
return b.String()
}
-func (slp *simpleLocationProvider) NewDataLocation(dataFileName string) string
{
- return joinLocationPath(slp.dataPath, dataFileName)
+func (slp *simpleLocationProvider) NewDataLocation(dataFileName, partitionPath
string) string {
+ if partitionPath == "" {
+ return joinLocationPath(slp.dataPath, dataFileName)
+ }
+
+ return joinLocationPath(slp.dataPath, partitionPath, dataFileName)
}
func (slp *simpleLocationProvider) NewTableMetadataFileLocation(newVersion
int) (string, error) {
@@ -151,17 +156,19 @@ func computeHash(dataFileName string) string {
return string(hashPath[:])
}
-func (p *objectStoreLocationProvider) NewDataLocation(dataFileName string)
string {
- if path.Dir(dataFileName) != "." {
- return p.simpleLocationProvider.NewDataLocation(dataFileName)
+// NewDataLocation places the entropy hash right under the data path, like the
+// Java implementation. The partition directories are kept, and hashed with the
+// file name, only when partitioned paths are enabled.
+func (p *objectStoreLocationProvider) NewDataLocation(dataFileName,
partitionPath string) string {
+ if !p.includePartitionPaths {
+ return joinLocationPath(p.dataPath,
computeHash(dataFileName)+"-"+dataFileName)
}
- hashedPath := computeHash(dataFileName)
- if p.includePartitionPaths {
- return joinLocationPath(p.dataPath, hashedPath, dataFileName)
- } else {
- return joinLocationPath(p.dataPath, hashedPath+"-"+dataFileName)
+ if partitionPath != "" {
+ dataFileName = partitionPath + "/" + dataFileName
}
+
+ return joinLocationPath(p.dataPath, computeHash(dataFileName),
dataFileName)
}
func newObjectStoreLocationProvider(tableLoc string, tableProps
iceberg.Properties) *objectStoreLocationProvider {
diff --git a/table/locations_test.go b/table/locations_test.go
index 5bdd4d1c3..e72e0a42a 100644
--- a/table/locations_test.go
+++ b/table/locations_test.go
@@ -57,8 +57,8 @@ func TestLocationProviderLocalPaths(t *testing.T) {
metadataFile, err :=
provider.NewTableMetadataFileLocation(1)
require.NoError(t, err)
for _, file := range []struct{
location, dir string }{
-
{provider.NewDataLocation("data.parquet"), dataDir},
-
{provider.NewDataLocation("partition=value/data.parquet"), dataDir},
+
{provider.NewDataLocation("data.parquet", ""), dataDir},
+
{provider.NewDataLocation("data.parquet", "partition=value"), dataDir},
{provider.NewMetadataLocation("manifest.avro"), metadataDir},
{metadataFile,
metadataDir},
} {
@@ -103,18 +103,18 @@ func TestLocationProviderLiteralPaths(t *testing.T) {
t.Run(tt.location, func(t *testing.T) {
provider, err :=
table.LoadLocationProvider(tt.location, nil)
require.NoError(t, err)
- assert.Equal(t, tt.data, provider.NewDataLocation("a"))
+ assert.Equal(t, tt.data, provider.NewDataLocation("a",
""))
assert.Equal(t, tt.metadata,
provider.NewMetadataLocation("manifest.avro"))
})
}
}
func TestLocationProviderLiteralCustomPaths(t *testing.T) {
- for _, tt := range []struct{ objectStore, partitionPaths, data string }{
- {"false", "false", "s3://bucket/my data%20#?/a"},
- {"false", "true", "s3://bucket/my data%20#?/a"},
- {"true", "false", "s3://bucket/my
data%20#?/0101/0110/1001/10110010-a"},
- {"true", "true", "s3://bucket/my
data%20#?/0101/0110/1001/10110010/a"},
+ for _, tt := range []struct{ objectStore, partitionPaths, data,
partitioned, multiLevel string }{
+ {"false", "false", "s3://bucket/my data%20#?/a",
"s3://bucket/my data%20#?/partition=a b%20#?/file.parquet", "s3://bucket/my
data%20#?/a=1/b=2/file.parquet"},
+ {"false", "true", "s3://bucket/my data%20#?/a", "s3://bucket/my
data%20#?/partition=a b%20#?/file.parquet", "s3://bucket/my
data%20#?/a=1/b=2/file.parquet"},
+ {"true", "false", "s3://bucket/my
data%20#?/0101/0110/1001/10110010-a", "s3://bucket/my
data%20#?/0010/1111/0101/11011101-file.parquet", "s3://bucket/my
data%20#?/0010/1111/0101/11011101-file.parquet"},
+ {"true", "true", "s3://bucket/my
data%20#?/0101/0110/1001/10110010/a", "s3://bucket/my
data%20#?/1011/0010/0000/11011010/partition=a b%20#?/file.parquet",
"s3://bucket/my data%20#?/1110/0111/1101/10000010/a=1/b=2/file.parquet"},
} {
t.Run(tt.objectStore+"/"+tt.partitionPaths, func(t *testing.T) {
provider, err :=
table.LoadLocationProvider("s3://bucket/my table", iceberg.Properties{
@@ -124,9 +124,10 @@ func TestLocationProviderLiteralCustomPaths(t *testing.T) {
table.WriteMetadataPathKey:
"gs://bucket/100% metadata/./../here/",
})
require.NoError(t, err)
- assert.Equal(t, tt.data, provider.NewDataLocation("a"))
- assert.Equal(t, "s3://bucket/my data%20#?/partition=a
b%20#?/file.parquet",
- provider.NewDataLocation("partition=a
b%20#?/file.parquet"))
+ assert.Equal(t, tt.data, provider.NewDataLocation("a",
""))
+ assert.Equal(t, tt.partitioned,
+ provider.NewDataLocation("file.parquet",
"partition=a b%20#?"))
+ assert.Equal(t, tt.multiLevel,
provider.NewDataLocation("file.parquet", "a=1/b=2"))
assert.Equal(t, "gs://bucket/100%
metadata/./../here/manifest #?%.avro",
provider.NewMetadataLocation("manifest
#?%.avro"))
})
@@ -185,12 +186,12 @@ func TestObjectStoreLocationProvider(t *testing.T) {
iceberg.Properties{table.ObjectStoreEnabledKey: "true"})
require.NoError(t, err)
- assert.Equal(t, "table_location/data/0101/0110/1001/10110010/a",
provider.NewDataLocation("a"))
- assert.Equal(t, "table_location/data/1110/0111/1110/00000011/b",
provider.NewDataLocation("b"))
- assert.Equal(t, "table_location/data/0010/1101/0110/01011111/c",
provider.NewDataLocation("c"))
- assert.Equal(t, "table_location/data/1001/0001/0100/01110011/d",
provider.NewDataLocation("d"))
+ assert.Equal(t, "table_location/data/0101/0110/1001/10110010/a",
provider.NewDataLocation("a", ""))
+ assert.Equal(t, "table_location/data/1110/0111/1110/00000011/b",
provider.NewDataLocation("b", ""))
+ assert.Equal(t, "table_location/data/0010/1101/0110/01011111/c",
provider.NewDataLocation("c", ""))
+ assert.Equal(t, "table_location/data/1001/0001/0100/01110011/d",
provider.NewDataLocation("d", ""))
assert.Equal(t,
"table_location/data/0000/0000/0000/00000000/part-00164929.parquet",
- provider.NewDataLocation("part-00164929.parquet"))
+ provider.NewDataLocation("part-00164929.parquet", ""))
}
// TestObjectStoreLocationProviderPartitionedPathsDisabled tests that when
partitioned paths are disabled,
@@ -201,9 +202,9 @@ func
TestObjectStoreLocationProviderPartitionedPathsDisabled(t *testing.T) {
iceberg.Properties{table.ObjectStoreEnabledKey: "true",
table.WriteObjectStorePartitionedPathsKey: "false"})
require.NoError(t, err)
- assert.Equal(t, "table_location/data/0101/0110/1001/10110010-a",
provider.NewDataLocation("a"))
- assert.Equal(t, "table_location/data/1110/0111/1110/00000011-b",
provider.NewDataLocation("b"))
- assert.Equal(t, "table_location/data/0010/1101/0110/01011111-c",
provider.NewDataLocation("c"))
- assert.Equal(t, "table_location/data/1001/0001/0100/01110011-d",
provider.NewDataLocation("d"))
- assert.Equal(t,
"table_location/data/0110/1010/0011/11101000-test.parquet",
provider.NewDataLocation("test.parquet"))
+ assert.Equal(t, "table_location/data/0101/0110/1001/10110010-a",
provider.NewDataLocation("a", ""))
+ assert.Equal(t, "table_location/data/1110/0111/1110/00000011-b",
provider.NewDataLocation("b", ""))
+ assert.Equal(t, "table_location/data/0010/1101/0110/01011111-c",
provider.NewDataLocation("c", ""))
+ assert.Equal(t, "table_location/data/1001/0001/0100/01110011-d",
provider.NewDataLocation("d", ""))
+ assert.Equal(t,
"table_location/data/0110/1010/0011/11101000-test.parquet",
provider.NewDataLocation("test.parquet", ""))
}
diff --git a/table/rolling_data_writer.go b/table/rolling_data_writer.go
index 507fde2f3..5c1801cb5 100644
--- a/table/rolling_data_writer.go
+++ b/table/rolling_data_writer.go
@@ -23,8 +23,6 @@ import (
"fmt"
"iter"
"log/slog"
- "maps"
- "net/url"
"strings"
"sync"
"sync/atomic"
@@ -54,8 +52,6 @@ const rollingDataWriterQueueCapacity = 64
// for different partitions, providing shared configuration and coordination
// across all writers in a partitioned write operation.
type writerFactory struct {
- rootLocation string
- rootURL *url.URL
fs iceio.WriteFileIO
writeUUID *uuid.UUID
taskSchema *iceberg.Schema
@@ -90,13 +86,12 @@ type writerFactory struct {
// capacity; non-positive uses rollingDataWriterQueueCapacity.
recordBufferSize int
- writers sync.Map
- partitionLocProviders sync.Map
- nextCount func() (int, bool)
- stopCount func()
- countMu sync.Mutex
- partitionIDCounter atomic.Int64
- mu sync.Mutex
+ writers sync.Map
+ nextCount func() (int, bool)
+ stopCount func()
+ countMu sync.Mutex
+ partitionIDCounter atomic.Int64
+ mu sync.Mutex
}
type writerFactoryOption func(*writerFactory) error
@@ -150,13 +145,6 @@ func withFactoryFileSchema(schema *iceberg.Schema)
writerFactoryOption {
func newWriterFactory(rootLocation string, args recordWritingArgs, meta
*MetadataBuilder, taskSchema *iceberg.Schema, targetFileSize int64, opts
...writerFactoryOption) (*writerFactory, error) {
nextCount, stopCount := iter.Pull(args.counter)
- rootURL, err := url.Parse(rootLocation)
- if err != nil {
- stopCount()
-
- return nil, err
- }
-
locProvider, err := LoadLocationProvider(rootLocation, meta.props)
if err != nil {
stopCount()
@@ -220,8 +208,6 @@ func newWriterFactory(rootLocation string, args
recordWritingArgs, meta *Metadat
}
f := &writerFactory{
- rootLocation: rootLocation,
- rootURL: rootURL,
fs: args.fs,
writeUUID: args.writeUUID,
taskSchema: taskSchema,
@@ -309,16 +295,7 @@ func (w *writerFactory) openFileWriter(ctx
context.Context, partitionPath string
fileName := dataFileName(*w.writeUUID, cnt, partitionID, fileCount,
strings.ToLower(string(w.fileFormat)))
- var filePath string
- if partitionPath != "" {
- partitionLoc, err := w.partitionLocProvider(partitionPath)
- if err != nil {
- return nil, err
- }
- filePath = partitionLoc.NewDataLocation(fileName)
- } else {
- filePath = w.locProvider.NewDataLocation(fileName)
- }
+ filePath := w.locProvider.NewDataLocation(fileName, partitionPath)
// SortOrderID is set only for presorted input: otherwise batches are
sorted
// individually (see resolveSortKeys), which does not sort the whole
file.
@@ -338,25 +315,6 @@ func (w *writerFactory) openFileWriter(ctx
context.Context, partitionPath string
}, arrowSchema)
}
-func (w *writerFactory) partitionLocProvider(partitionPath string)
(LocationProvider, error) {
- if cached, ok := w.partitionLocProviders.Load(partitionPath); ok {
- return cached.(LocationProvider), nil
- }
-
- partitionDataPath := w.rootURL.JoinPath("data", partitionPath).String()
- partitionProps := make(iceberg.Properties, len(w.tableProps)+1)
- maps.Copy(partitionProps, w.tableProps)
- partitionProps[WriteDataPathKey] = partitionDataPath
- loc, err := LoadLocationProvider(w.rootLocation, partitionProps)
- if err != nil {
- return nil, err
- }
-
- w.partitionLocProviders.Store(partitionPath, loc)
-
- return loc, nil
-}
-
// RollingDataWriter writes Arrow records for a specific partition, rolling to
// new data files when the actual compressed file size reaches the target.
type RollingDataWriter struct {
diff --git a/table/rolling_data_writer_test.go
b/table/rolling_data_writer_test.go
index 717f6e5a2..15f5e0aec 100644
--- a/table/rolling_data_writer_test.go
+++ b/table/rolling_data_writer_test.go
@@ -20,8 +20,10 @@ package table
import (
"context"
"errors"
+ "maps"
"os"
"path/filepath"
+ "regexp"
"runtime"
"strconv"
"sync"
@@ -655,12 +657,11 @@ func (s *RollingDataWriterTestSuite)
TestBytesWrittenReflectsCompressedSize() {
s.Equal(int64(100), df.Count())
}
-// TestPartitionLocProviderPreservesObjectStorageProps reproduces #1155: when
-// write.object-storage.enabled is true on a partitioned table, partition data
-// file paths must include entropy hash dirs from objectStoreLocationProvider.
-// Previously partitionLocProvider rebuilt props with only WriteDataPathKey, so
-// LoadLocationProvider fell back to simpleLocationProvider for partitions.
-func (s *RollingDataWriterTestSuite)
TestPartitionLocProviderPreservesObjectStorageProps() {
+// TestPartitionedDataLocations checks partitioned data file paths against the
+// Java layout: with object storage the entropy hash comes right after the
+// data path, and write.object-storage.partitioned-paths=false drops the
+// partition directories (#2367). write.data.path is honored as well.
+func (s *RollingDataWriterTestSuite) TestPartitionedDataLocations() {
arrSchema := arrow.NewSchema([]arrow.Field{
{Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
{Name: "name", Type: arrow.BinaryTypes.String, Nullable: true},
@@ -669,42 +670,74 @@ func (s *RollingDataWriterTestSuite)
TestPartitionLocProviderPreservesObjectStor
icebergSchema, err := ArrowSchemaToIcebergWithFreshIDs(arrSchema, false)
s.Require().NoError(err)
- spec := iceberg.NewPartitionSpec()
- loc := "table_location"
- meta, err := NewMetadata(icebergSchema, &spec, UnsortedSortOrder, loc,
iceberg.Properties{
- ObjectStoreEnabledKey: "true",
- })
- s.Require().NoError(err)
- metaBuilder, err := MetadataBuilderFromBase(meta, "")
- s.Require().NoError(err)
+ const (
+ partition = "dt=2026-06-04"
+ hash = `[01]{4}/[01]{4}/[01]{4}/[01]{8}`
+ file = `[^/]+\.parquet`
+ )
- writeUUID := uuid.New()
- args := recordWritingArgs{
- sc: arrSchema,
- fs: iceio.LocalFS{},
- writeUUID: &writeUUID,
- counter: func(yield func(int) bool) {
- for i := 0; ; i++ {
- if !yield(i) {
- break
- }
+ for _, tt := range []struct {
+ name string
+ props iceberg.Properties
+ want string
+ }{
+ {"simple", iceberg.Properties{}, "data/" + partition + "/" +
file},
+ {"simple custom path", iceberg.Properties{WriteDataPathKey:
"custom"}, "custom/" + partition + "/" + file},
+ {"object storage", iceberg.Properties{ObjectStoreEnabledKey:
"true"}, "data/" + hash + "/" + partition + "/" + file},
+ {"object storage without partition paths", iceberg.Properties{
+ ObjectStoreEnabledKey: "true",
+ WriteObjectStorePartitionedPathsKey: "false",
+ }, "data/" + hash + "-" + file},
+ {"object storage custom path", iceberg.Properties{
+ ObjectStoreEnabledKey: "true",
+ WriteDataPathKey: "custom",
+ }, "custom/" + hash + "/" + partition + "/" + file},
+ } {
+ s.Run(tt.name, func() {
+ loc := filepath.ToSlash(s.T().TempDir())
+ props := maps.Clone(tt.props)
+ if dataPath, ok := props[WriteDataPathKey]; ok {
+ props[WriteDataPathKey] = loc + "/" + dataPath
}
- },
- }
- factory, err := newWriterFactory(loc, args, metaBuilder, icebergSchema,
1024*1024)
- s.Require().NoError(err)
- defer factory.closeAll()
- provider, err := factory.partitionLocProvider("dt=2026-06-04")
- s.Require().NoError(err)
- _, ok := provider.(*objectStoreLocationProvider)
- s.True(ok, "expected objectStoreLocationProvider for partitioned writes
when write.object-storage.enabled=true")
-
- // Hashed prefix should appear before the file name for object storage
layout.
- dataLoc := provider.NewDataLocation("a")
- s.Contains(dataLoc, "table_location/data/dt=2026-06-04/")
- s.NotEqual("table_location/data/dt=2026-06-04/a", dataLoc,
- "object storage layout should inject an entropy hash prefix
between partition dir and file")
+ spec := iceberg.NewPartitionSpec()
+ meta, err := NewMetadata(icebergSchema, &spec,
UnsortedSortOrder, loc, props)
+ s.Require().NoError(err)
+ metaBuilder, err := MetadataBuilderFromBase(meta, "")
+ s.Require().NoError(err)
+
+ writeUUID := uuid.New()
+ args := recordWritingArgs{
+ sc: arrSchema,
+ fs: iceio.LocalFS{},
+ writeUUID: &writeUUID,
+ counter: func(yield func(int) bool) {
+ for i := 0; ; i++ {
+ if !yield(i) {
+ break
+ }
+ }
+ },
+ }
+ factory, err := newWriterFactory(loc, args,
metaBuilder, icebergSchema, 1024*1024)
+ s.Require().NoError(err)
+ defer factory.closeAll()
+
+ outputCh := make(chan iceberg.DataFile, 1)
+ writer := factory.newRollingDataWriter(s.ctx,
partition, nil, outputCh)
+
+ record := s.buildRecord(arrSchema, 5)
+ defer record.Release()
+
+ s.Require().NoError(writer.Add(record))
+ s.Require().NoError(writer.closeAndWait())
+ close(outputCh)
+
+ df := <-outputCh
+ s.Require().NotNil(df)
+ s.Regexp("^"+regexp.QuoteMeta(loc)+"/"+tt.want+"$",
df.FilePath())
+ })
+ }
}
// TestSortsRowsBeforeWriting exercises the sort-on-write path: the rolling
diff --git a/table/writer.go b/table/writer.go
index 16b32cefa..98405b9c9 100644
--- a/table/writer.go
+++ b/table/writer.go
@@ -218,7 +218,7 @@ func (w *defaultDataFileWriter) writeFile(ctx
context.Context, partitionValues m
}
filePath := w.loc.NewDataLocation(
- task.GenerateDataFileName(w.invariants.extension))
+ task.GenerateDataFileName(w.invariants.extension), "")
return w.format.WriteDataFile(ctx, w.fs, partitionValues,
tblutils.WriteFileInfo{
FileSchema: w.fileSchema,
diff --git a/website/src/releases.md b/website/src/releases.md
index dd9e3eda6..0dc5a7c51 100644
--- a/website/src/releases.md
+++ b/website/src/releases.md
@@ -42,6 +42,8 @@ Per-release notes (highlights, breaking changes,
contributors) are published on
### Unreleased compatibility notes
+- **`LocationProvider.NewDataLocation` takes the partition path.** The method
is now `NewDataLocation(dataFileName, partitionPath string)`, matching Iceberg
Java and PyIceberg, so custom `LocationProvider` implementations must add the
argument. Callers pass the partition path there, or `""` for an unpartitioned
file, instead of prefixing the file name.
+- **Partitioned data files follow the table's location provider.** With
`write.object-storage.enabled=true` and the default
`write.object-storage.partitioned-paths=true`, new partitioned files land under
`data/<hash>/<partition>/` instead of `data/<partition>/<hash>/`, which can
affect prefix-based lifecycle rules and bucket policies.
`write.object-storage.partitioned-paths=false` and `write.data.path` now apply
to partitioned writes. Partitioned paths also no longer percent-encode the t
[...]
- **Catalog capabilities split into narrower interfaces.** `catalog.Catalog`
now embeds `TableCatalog` and `NamespaceCatalog` with unchanged table and
namespace methods. `Name() string` is now required, breaking third-party
catalogs and test fakes. `ViewCatalog` is optional (REST and Hive), while REST
also implements `UpdatableViewCatalog` for `UpdateView`. Use
`catalog.As[T](cat)` and the `Unwrap() Catalog` convention to discover optional
capabilities through decorators. Future capabili [...]
- **`ReplaceFilesWithDeleteFiles` removed.** For delete-file replacements, use
`tx.NewRewrite(snapshotProps)`, call `DeleteFile` for each source file and
`AddDeleteFileWithDataSequenceNumber` for each replacement, then commit the
builder. Rewrites record a `replace` snapshot instead of `overwrite`. The
five-argument `ReplaceFiles` API is unchanged and remains available for
compaction and delete-file removals.
- **Path-based data-file addition renamed.** Replace `Transaction.AddFiles`
with `AddDataFilesByPath`, `AddFilesOption` with `AddDataFilesByPathOption`,
and `WithAddFilesConcurrency` with `WithAddDataFilesByPathConcurrency`.
Behavior is unchanged.
