You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "mzzz-zzm (via GitHub)" <gi...@apache.org> on 2026/05/04 22:18:54 UTC
[PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
mzzz-zzm opened a new pull request, #983:
URL: https://github.com/apache/iceberg-go/pull/983
Fixes #978
## Problem
`RowDelta.validate` passes `iceberg.AlwaysTrue{}` to `validateNoConflictingDataFiles` whenever equality-delete files are present. This means any concurrent append to **any** partition is treated as a conflict, even when it lands in a completely different partition from the equality-deletes. Under serializable isolation this causes spurious `ErrConflictingDataFiles` errors for workloads that write to multiple independent partitions concurrently.
## Fix
`RowDelta` now collects the partition tuples of all equality-delete files it adds (`eqDeletePartitions`). A new validator, `validateNoConflictingDataFilesInPartitions`, checks concurrent data files only in those specific partition tuples:
- If the partition set is empty (no eq-deletes), the check is skipped.
- If any eq-delete is unpartitioned (empty tuple), it falls back to the conservative `AlwaysTrue` check, preserving existing safety.
- Otherwise, only concurrent files in a matching partition are flagged.
## Files changed
- `table/row_delta.go`: collect `eqDeletePartitions` instead of `hasEqDeletes`
- `table/conflict_validation.go`: `validateNoConflictingDataFilesInPartitions` + `partitionTupleKey`
- `table/partition_conflict_test.go`: unit tests for both new functions
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3195690281
##########
table/conflict_validation.go:
##########
@@ -426,6 +427,159 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to the partitions touched
+// by equality-delete files. It builds an OR-of-equalities filter from
+// the provided partition tuples and delegates to
+// validateAddedDataFilesMatchingFilter, which performs per-spec
+// projection, manifest-summary pruning, and type-aware evaluation via
+// iceberg.Literal — making it safe for UUID, decimal, binary, fixed,
+// and future partition types, and correct across partition-spec
+// evolution because each concurrent manifest is projected against its
+// own spec ID.
+//
+// Callers are responsible for ensuring the table is partitioned
+// (i.e. at least one partition field exists) before calling this
+// function. For unpartitioned tables, call
+// validateNoConflictingDataFiles(ctx, iceberg.AlwaysTrue{}, level)
+// directly.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeleteFiles []iceberg.DataFile, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeleteFiles) == 0 {
+ return nil
+ }
+
+ filter, err := eqDeletePartitionsToFilter(eqDeleteFiles, ctx.current)
+ if err != nil {
+ return fmt.Errorf("building partition conflict filter: %w", err)
+ }
+
+ return validateNoConflictingDataFiles(ctx, filter, level)
+}
+
+// eqDeletePartitionsToFilter converts equality-delete data files into an
+// OR-of-ANDs BooleanExpression in row (source) space, suitable for passing
+// to validateAddedDataFilesMatchingFilter.
+//
+// For each eq-delete file it resolves each partition field ID to the source
+// schema field name via the file's partition spec, then builds an EqualTo
+// predicate using Reference(sourceFieldName). Multiple fields within one
+// partition are AND-ed; multiple eq-delete files are OR-ed.
+//
+// The resulting expression is projected per-concurrent-manifest's spec ID
+// inside validateAddedDataFilesMatchingFilter (via buildPartitionProjection),
+// ensuring correct conflict detection even after partition-spec evolution.
+//
+// An empty partition tuple (unpartitioned delete) returns AlwaysTrue so the
+// caller falls back to the conservative full-table scan. Callers should
+// normally guard against calling this function for unpartitioned tables (see
+// RowDelta.validate).
+func eqDeletePartitionsToFilter(files []iceberg.DataFile, meta Metadata) (iceberg.BooleanExpression, error) {
+ terms := make([]iceberg.BooleanExpression, 0, len(files))
+ for _, f := range files {
+ p := f.Partition()
+ if len(p) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ spec := meta.PartitionSpecByID(int(f.SpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("partition spec ID %d not found in metadata", f.SpecID())
+ }
+
+ // Build partition field ID → PartitionField lookup for this spec.
+ partFieldByID := make(map[int]iceberg.PartitionField, spec.NumFields())
+ for _, pf := range spec.Fields() {
+ partFieldByID[pf.FieldID] = pf
+ }
+
+ // Sort partition field IDs for deterministic expression order.
+ fieldIDs := make([]int, 0, len(p))
+ for id := range p {
+ fieldIDs = append(fieldIDs, id)
+ }
+ sort.Ints(fieldIDs)
+
+ conjuncts := make([]iceberg.BooleanExpression, 0, len(p))
+ for _, partFieldID := range fieldIDs {
+ pf, ok := partFieldByID[partFieldID]
+ if !ok {
+ return nil, fmt.Errorf("partition field ID %d not found in spec %d", partFieldID, f.SpecID())
+ }
+
+ // Resolve to source schema field to obtain the Reference name.
+ sourceField, ok := meta.CurrentSchema().FindFieldByID(pf.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("source field ID %d (partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ }
+
+ lit, err := anyToLiteral(p[partFieldID])
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q: %w", sourceField.Name, err)
+ }
+
+ conjuncts = append(conjuncts, iceberg.LiteralPredicate(iceberg.OpEQ, iceberg.Reference(sourceField.Name), lit))
Review Comment:
Fixed. `eqDeletePartitionsToFilter` now detects non-identity transforms per partition field before building any row-space predicate. If any field in a file's partition spec uses a non-identity transform (bucket, day, hour, truncate, year, month), the function falls back to `AlwaysTrue{}` for that file — treating the eq-delete as table-wide, which is conservative but always correct. `IdentityTransform` continues to produce the scoped row-space predicate as before.
The full `PartitionSet`-style fix (partition-space predicate keyed by `(specID, partitionFieldName)` evaluated only against same-spec concurrent manifests) is deferred to a follow-up PR.
Two regression tests lock the contract:
- `TestRowDeltaValidate_BucketTransformFallsBackToAlwaysTrue`: `bucket[16](user_id)` partitioned table — concurrent data in bucket 1, eq-delete in bucket 1 → rejected.
- `TestRowDeltaValidate_DayTransformFallsBackToAlwaysTrue`: `day(event_ts)` partitioned table — concurrent data in day 100, eq-delete in day 200 → still rejected (conservative; AlwaysTrue cannot distinguish days).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3188460747
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
+ return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
+ }
+ }
+
+ // Build a set of partition tuple keys for O(1) lookup.
+ partSet := make(map[string]struct{}, len(eqDeletePartitions))
+ for _, p := range eqDeletePartitions {
+ partSet[partitionTupleKey(p)] = struct{}{}
+ }
+
+ for _, snap := range ctx.concurrent {
+ manifests, err := snap.Manifests(ctx.fs)
+ if err != nil {
+ return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ entries, err := mf.FetchEntries(ctx.fs, false)
+ if err != nil {
+ return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
+ }
+ for _, e := range entries {
+ if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
+ continue
+ }
+ if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
+ return fmt.Errorf("%w: snapshot %d added data file %s in eq-delete partition",
+ ErrConflictingDataFiles, snap.SnapshotID, e.DataFile().FilePath())
+ }
+ }
+ }
+ }
+
+ return nil
+}
+
+// partitionTupleKey returns a deterministic string key for a partition
+// tuple map. Keys are sorted by field id so maps with the same content
+// always produce the same string.
+func partitionTupleKey(p map[int]any) string {
+ if len(p) == 0 {
+ return ""
+ }
+ keys := make([]int, 0, len(p))
+ for k := range p {
+ keys = append(keys, k)
+ }
+ sort.Ints(keys)
+ var buf []byte
+ for _, k := range keys {
+ buf = fmt.Appendf(buf, "%d=%v;", k, p[k])
Review Comment:
`fmt.Appendf("%d=%v;", k, p[k])` is pretty fragile as an equality oracle here. `map[int]any` partition values can land as several Go types depending on construction path — `uuid.UUID` (= `[16]byte`) vs raw `[16]byte`, `iceberg.DecimalLiteral` vs `*big.Rat`, `time.Time` with monotonic-clock suffix, etc., see `convertAvroValueToIcebergType` in `manifest.go:1756`. Same logical value, different `%v`, missed match. Also `=` and `;` aren't escaped, so a string value `"1=a;2=b"` collides with a different two-field tuple.
A silent miss here is worse than the bug being fixed — we'd accept a commit Java would reject. I'd either build an `Or(And(...))` partition expression and call the existing `validateAddedDataFilesMatchingFilter` (which uses `iceberg.Literal` comparison and handles all of these), or normalize each known type to a canonical hashable form before keying.
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
+ return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
+ }
+ }
+
+ // Build a set of partition tuple keys for O(1) lookup.
+ partSet := make(map[string]struct{}, len(eqDeletePartitions))
+ for _, p := range eqDeletePartitions {
+ partSet[partitionTupleKey(p)] = struct{}{}
+ }
+
+ for _, snap := range ctx.concurrent {
+ manifests, err := snap.Manifests(ctx.fs)
+ if err != nil {
+ return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ entries, err := mf.FetchEntries(ctx.fs, false)
+ if err != nil {
+ return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
+ }
+ for _, e := range entries {
+ if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
+ continue
+ }
+ if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
Review Comment:
The lookup compares only `(field_id, value)` and never consults `mf.PartitionSpecID()`. After a partition-spec evolution (renamed field, identity → bucket, day → hour), the same logical row gets written under a different tuple — eq-delete bound to spec A might be `{1: "hot"}` while concurrent data under spec B is `{1: 3}` (bucketed) or `{2: "hot"}` (renamed). Tuples never match, real conflict missed.
The sibling helper at lines 343-434 handles this via `buildPartitionProjection(specID, ...)` keyed per spec id. I'd at minimum key on `(specID, tuple)` and use `mf.PartitionSpecID()` plus the eq-delete file's `SpecID()`. Routing through `validateAddedDataFilesMatchingFilter` with an OR-of-equalities filter gets it for free.
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
Review Comment:
Bigger-picture: I'd consider folding this into `validateAddedDataFilesMatchingFilter` rather than adding a sibling. Derive a `BooleanExpression` from `eqDeletePartitions` like `Or(And(field_a == v_a, field_b == v_b), ...)` and pass it to `validateNoConflictingDataFiles(ctx, filter, level)`. That helper already does per-spec projection (spec-evolution correct), manifest-summary pruning (no full-manifest scan when summaries can't match), and type-aware partition evaluation via `iceberg.Literal` (no UUID/decimal/binary issues).
Resolves the comments on `partitionTupleKey` and spec-id awareness at the same time, and matches the pattern documented in this file's preamble ("there is one code path that pruning semantics flow through") and Java's `MergingSnapshotProducer.validateNoNewDataFiles`.
##########
table/partition_conflict_test.go:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+package table
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// ---------------------------------------------------------------------------
+// partitionTupleKey
+// ---------------------------------------------------------------------------
+
+func TestPartitionTupleKey_Empty(t *testing.T) {
+ assert.Equal(t, "", partitionTupleKey(map[int]any{}))
+ assert.Equal(t, "", partitionTupleKey(nil))
+}
+
+func TestPartitionTupleKey_SingleField(t *testing.T) {
+ assert.Equal(t, "1=hello;", partitionTupleKey(map[int]any{1: "hello"}))
+}
+
+func TestPartitionTupleKey_MultipleFieldsOrderedByID(t *testing.T) {
+ // Keys must be sorted by field ID, so insertion order must not matter.
+ p1 := map[int]any{1: "a", 3: "c", 2: "b"}
+ p2 := map[int]any{3: "c", 1: "a", 2: "b"}
+ assert.Equal(t, partitionTupleKey(p1), partitionTupleKey(p2),
+ "same content, different insertion order must produce identical keys")
+ assert.Equal(t, "1=a;2=b;3=c;", partitionTupleKey(p1))
+}
+
+func TestPartitionTupleKey_DifferentValuesProduceDifferentKeys(t *testing.T) {
+ pA := map[int]any{1: "us-east-1"}
+ pB := map[int]any{1: "eu-west-1"}
+ assert.NotEqual(t, partitionTupleKey(pA), partitionTupleKey(pB))
+}
+
+// ---------------------------------------------------------------------------
+// validateNoConflictingDataFilesInPartitions — short-circuit paths
+// ---------------------------------------------------------------------------
+
+// TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp
+// verifies that the validator is a no-op under snapshot isolation, matching
+// the behaviour of validateNoConflictingDataFiles.
+func TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ partitions := []map[int]any{{1: "us-east-1"}}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, partitions, IsolationSnapshot))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_EmptyInputs verifies that
+// the validator short-circuits when there are no concurrent snapshots or no
+// eq-delete partitions, without touching the filesystem.
+func TestValidateNoConflictingDataFilesInPartitions_EmptyInputs(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // No concurrent snapshots — must not read any manifests.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{{1: "us-east-1"}}, IsolationSerializable))
+
+ // No eq-delete partitions — nothing to validate.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, nil, IsolationSerializable))
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{}, IsolationSerializable))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack
+// verifies that a single unpartitioned eq-delete (empty tuple) causes the
+// validator to fall back to the conservative AlwaysTrue path. Because the
+// conflictContext has no concurrent snapshots (fs = nil, concurrent = nil),
+// the AlwaysTrue fallback itself returns nil — this test just ensures we
+// reach and exercise that code path without panicking.
+func TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack(t *testing.T) {
+ head := int64(1)
+ // Build metadata with a concurrent snapshot by using two different base/current metas.
+ meta := newConflictTestMetadata(t, &head)
+
+ // writerBase has no snapshot; current has head snapshot → head is concurrent.
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // Empty partition tuple → unpartitioned eq-delete → must fall back to AlwaysTrue.
+ // With no concurrent snapshots in ctx the AlwaysTrue path returns nil (no-op).
+ emptyPartition := map[int]any{}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{emptyPartition}, IsolationSerializable))
+}
Review Comment:
The #978 reproducer (`TestBugRepro_RowDeltaFalseConflictDifferentPartition` from the issue body) is the test that would fail on `main` and pass on this branch — it's the regression guard for the actual contract of this PR, and it isn't here.
A few I'd want at minimum:
* the #978 reproducer as-is — different-partition allowed
* same-partition rejected — without this, a future change to `partitionTupleKey` could silently degrade the validator into a no-op with no signal
* UUID or decimal partition column, same-partition rejected — would fail today against `%v` keying, that's the point
* spec evolution — eq-delete bound to spec A, concurrent data under spec B (renamed field or transform change), will fail today
* a genuinely-unpartitioned table replacing the trivial-pass test above
The harness exists — `conflict_validation_test.go` builds metadata with concurrent snapshots and `row_delta_test.go` builds eq-delete files. They just need to be wired together.
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
Review Comment:
I'd hoist this fallback to the caller. The function name says "InPartitions", but a single empty input element silently turns it into AlwaysTrue across the whole table — leaky. Also "empty per-file partition tuple" is a noisy proxy for "unpartitioned table": an eq-delete file with an unset partition map on a *partitioned* table (easy to do via `NewDataFileBuilder` without partition data) silently nullifies the optimization. The right signal is the spec itself.
Move the decision into `RowDelta.validate`: if the table's only spec is unpartitioned, call the AlwaysTrue version directly; otherwise the partition-scoped one. Both paths become explicit.
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
+ return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
+ }
+ }
+
+ // Build a set of partition tuple keys for O(1) lookup.
+ partSet := make(map[string]struct{}, len(eqDeletePartitions))
+ for _, p := range eqDeletePartitions {
+ partSet[partitionTupleKey(p)] = struct{}{}
+ }
+
+ for _, snap := range ctx.concurrent {
+ manifests, err := snap.Manifests(ctx.fs)
+ if err != nil {
+ return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ entries, err := mf.FetchEntries(ctx.fs, false)
+ if err != nil {
+ return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
+ }
+ for _, e := range entries {
+ if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
+ continue
+ }
+ if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
+ return fmt.Errorf("%w: snapshot %d added data file %s in eq-delete partition",
Review Comment:
The sibling on line 404 is `"snapshot %d added data file %s matching filter %s"` — includes the filter that triggered the match. This one says "in eq-delete partition" without saying which one. With multiple eq-delete files spanning multiple partitions, an operator triaging this can't tell which one conflicted without reading manifests by hand.
```go
return fmt.Errorf("%w: snapshot %d added data file %s in partition %v overlapping eq-delete",
ErrConflictingDataFiles, snap.SnapshotID, e.DataFile().FilePath(), e.DataFile().Partition())
```
##########
table/partition_conflict_test.go:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+package table
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// ---------------------------------------------------------------------------
+// partitionTupleKey
+// ---------------------------------------------------------------------------
+
+func TestPartitionTupleKey_Empty(t *testing.T) {
+ assert.Equal(t, "", partitionTupleKey(map[int]any{}))
+ assert.Equal(t, "", partitionTupleKey(nil))
+}
+
+func TestPartitionTupleKey_SingleField(t *testing.T) {
+ assert.Equal(t, "1=hello;", partitionTupleKey(map[int]any{1: "hello"}))
+}
+
+func TestPartitionTupleKey_MultipleFieldsOrderedByID(t *testing.T) {
+ // Keys must be sorted by field ID, so insertion order must not matter.
+ p1 := map[int]any{1: "a", 3: "c", 2: "b"}
+ p2 := map[int]any{3: "c", 1: "a", 2: "b"}
+ assert.Equal(t, partitionTupleKey(p1), partitionTupleKey(p2),
+ "same content, different insertion order must produce identical keys")
+ assert.Equal(t, "1=a;2=b;3=c;", partitionTupleKey(p1))
+}
+
+func TestPartitionTupleKey_DifferentValuesProduceDifferentKeys(t *testing.T) {
+ pA := map[int]any{1: "us-east-1"}
+ pB := map[int]any{1: "eu-west-1"}
+ assert.NotEqual(t, partitionTupleKey(pA), partitionTupleKey(pB))
+}
+
+// ---------------------------------------------------------------------------
+// validateNoConflictingDataFilesInPartitions — short-circuit paths
+// ---------------------------------------------------------------------------
+
+// TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp
+// verifies that the validator is a no-op under snapshot isolation, matching
+// the behaviour of validateNoConflictingDataFiles.
+func TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ partitions := []map[int]any{{1: "us-east-1"}}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, partitions, IsolationSnapshot))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_EmptyInputs verifies that
+// the validator short-circuits when there are no concurrent snapshots or no
+// eq-delete partitions, without touching the filesystem.
+func TestValidateNoConflictingDataFilesInPartitions_EmptyInputs(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // No concurrent snapshots — must not read any manifests.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{{1: "us-east-1"}}, IsolationSerializable))
+
+ // No eq-delete partitions — nothing to validate.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, nil, IsolationSerializable))
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{}, IsolationSerializable))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack
+// verifies that a single unpartitioned eq-delete (empty tuple) causes the
+// validator to fall back to the conservative AlwaysTrue path. Because the
+// conflictContext has no concurrent snapshots (fs = nil, concurrent = nil),
+// the AlwaysTrue fallback itself returns nil — this test just ensures we
+// reach and exercise that code path without panicking.
+func TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack(t *testing.T) {
+ head := int64(1)
+ // Build metadata with a concurrent snapshot by using two different base/current metas.
+ meta := newConflictTestMetadata(t, &head)
+
+ // writerBase has no snapshot; current has head snapshot → head is concurrent.
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // Empty partition tuple → unpartitioned eq-delete → must fall back to AlwaysTrue.
+ // With no concurrent snapshots in ctx the AlwaysTrue path returns nil (no-op).
+ emptyPartition := map[int]any{}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{emptyPartition}, IsolationSerializable))
Review Comment:
This test doesn't actually exercise the fallback. `newConflictContext(meta, meta, ...)` produces `ctx.concurrent = []` because base and current point at the same head, so the validator short-circuits at `len(ctx.concurrent) == 0` before reaching the empty-tuple loop. The comment in the test admits it: *"With no concurrent snapshots in ctx the AlwaysTrue path returns nil (no-op)"* — it'd still pass if we deleted the fallback entirely.
I'd build a context with a real concurrent snapshot (mirror `TestNewConflictContext_WriterHasNoBranchView` — `base = newConflictTestMetadata(t, nil)`, `current = newConflictTestMetadata(t, &head)`), then assert the fallback is reached, either by injecting a fake `iceio.IO` and observing the manifest fetch, or by an end-to-end fixture where AlwaysTrue would correctly flag a conflict the partition-scoped check would miss.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193720629
##########
table/partition_conflict_test.go:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+package table
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// ---------------------------------------------------------------------------
+// partitionTupleKey
+// ---------------------------------------------------------------------------
+
+func TestPartitionTupleKey_Empty(t *testing.T) {
+ assert.Equal(t, "", partitionTupleKey(map[int]any{}))
+ assert.Equal(t, "", partitionTupleKey(nil))
+}
+
+func TestPartitionTupleKey_SingleField(t *testing.T) {
+ assert.Equal(t, "1=hello;", partitionTupleKey(map[int]any{1: "hello"}))
+}
+
+func TestPartitionTupleKey_MultipleFieldsOrderedByID(t *testing.T) {
+ // Keys must be sorted by field ID, so insertion order must not matter.
+ p1 := map[int]any{1: "a", 3: "c", 2: "b"}
+ p2 := map[int]any{3: "c", 1: "a", 2: "b"}
+ assert.Equal(t, partitionTupleKey(p1), partitionTupleKey(p2),
+ "same content, different insertion order must produce identical keys")
+ assert.Equal(t, "1=a;2=b;3=c;", partitionTupleKey(p1))
+}
+
+func TestPartitionTupleKey_DifferentValuesProduceDifferentKeys(t *testing.T) {
+ pA := map[int]any{1: "us-east-1"}
+ pB := map[int]any{1: "eu-west-1"}
+ assert.NotEqual(t, partitionTupleKey(pA), partitionTupleKey(pB))
+}
+
+// ---------------------------------------------------------------------------
+// validateNoConflictingDataFilesInPartitions — short-circuit paths
+// ---------------------------------------------------------------------------
+
+// TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp
+// verifies that the validator is a no-op under snapshot isolation, matching
+// the behaviour of validateNoConflictingDataFiles.
+func TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ partitions := []map[int]any{{1: "us-east-1"}}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, partitions, IsolationSnapshot))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_EmptyInputs verifies that
+// the validator short-circuits when there are no concurrent snapshots or no
+// eq-delete partitions, without touching the filesystem.
+func TestValidateNoConflictingDataFilesInPartitions_EmptyInputs(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // No concurrent snapshots — must not read any manifests.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{{1: "us-east-1"}}, IsolationSerializable))
+
+ // No eq-delete partitions — nothing to validate.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, nil, IsolationSerializable))
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{}, IsolationSerializable))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack
+// verifies that a single unpartitioned eq-delete (empty tuple) causes the
+// validator to fall back to the conservative AlwaysTrue path. Because the
+// conflictContext has no concurrent snapshots (fs = nil, concurrent = nil),
+// the AlwaysTrue fallback itself returns nil — this test just ensures we
+// reach and exercise that code path without panicking.
+func TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack(t *testing.T) {
+ head := int64(1)
+ // Build metadata with a concurrent snapshot by using two different base/current metas.
+ meta := newConflictTestMetadata(t, &head)
+
+ // writerBase has no snapshot; current has head snapshot → head is concurrent.
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // Empty partition tuple → unpartitioned eq-delete → must fall back to AlwaysTrue.
+ // With no concurrent snapshots in ctx the AlwaysTrue path returns nil (no-op).
+ emptyPartition := map[int]any{}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{emptyPartition}, IsolationSerializable))
+}
Review Comment:
All five requested cases are now present in `table/partition_conflict_test.go`:
- `TestRowDeltaValidate_DifferentPartitionAllowed` — #978 reproducer: `eu-west-1` concurrent + `us-east-1` eq-delete → no conflict
- `TestRowDeltaValidate_SamePartitionRejected` — `us-east-1` concurrent + `us-east-1` eq-delete → rejected
- `TestRowDeltaValidate_UUIDPartitionSameRejected` / `TestRowDeltaValidate_UUIDPartitionDifferentAllowed` — UUID partition type safety (these would fail against `%v` keying)
- `TestRowDeltaValidate_UnpartitionedTableFallsBackToAlwaysTrue` — replaces the trivial-pass test
- `TestRowDeltaValidate_SpecEvolutionConflictDetected` — eq-delete written under spec A (`identity(region)`, partitionFieldID=1000), concurrent data written under spec B (renamed to `region_v2`, partitionFieldID=1001, same source field ID=2), both with value `"us-east-1"`. `eqDeletePartitionsToFilter` builds `Reference("region") == "us-east-1"` in row space; `validateAddedDataFilesMatchingFilter` projects it against spec B via `buildPartitionProjection` → conflict detected.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193622630
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
+ return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
+ }
+ }
+
+ // Build a set of partition tuple keys for O(1) lookup.
+ partSet := make(map[string]struct{}, len(eqDeletePartitions))
+ for _, p := range eqDeletePartitions {
+ partSet[partitionTupleKey(p)] = struct{}{}
+ }
+
+ for _, snap := range ctx.concurrent {
+ manifests, err := snap.Manifests(ctx.fs)
+ if err != nil {
+ return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ entries, err := mf.FetchEntries(ctx.fs, false)
+ if err != nil {
+ return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
+ }
+ for _, e := range entries {
+ if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
+ continue
+ }
+ if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
+ return fmt.Errorf("%w: snapshot %d added data file %s in eq-delete partition",
+ ErrConflictingDataFiles, snap.SnapshotID, e.DataFile().FilePath())
+ }
+ }
+ }
+ }
+
+ return nil
+}
+
+// partitionTupleKey returns a deterministic string key for a partition
+// tuple map. Keys are sorted by field id so maps with the same content
+// always produce the same string.
+func partitionTupleKey(p map[int]any) string {
+ if len(p) == 0 {
+ return ""
+ }
+ keys := make([]int, 0, len(p))
+ for k := range p {
+ keys = append(keys, k)
+ }
+ sort.Ints(keys)
+ var buf []byte
+ for _, k := range keys {
+ buf = fmt.Appendf(buf, "%d=%v;", k, p[k])
Review Comment:
`partitionTupleKey` has been removed entirely. The new `anyToLiteral` helper uses a type switch over all `iceberg.LiteralType` values — `bool`, `int32`, `int64`, `float32`, `float64`, `string`, `[]byte`, `Date`, `Time`, `Timestamp`, `TimestampNano`, `Decimal`, `uuid.UUID` — to produce typed `iceberg.Literal` values with stable equality semantics. No more `fmt.Sprintf("%v")` instability for UUID/decimal/binary. The `=`/`;` injection issue is gone — string-keyed maps are no longer used at all.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193629955
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
+ return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
+ }
+ }
+
+ // Build a set of partition tuple keys for O(1) lookup.
+ partSet := make(map[string]struct{}, len(eqDeletePartitions))
+ for _, p := range eqDeletePartitions {
+ partSet[partitionTupleKey(p)] = struct{}{}
+ }
+
+ for _, snap := range ctx.concurrent {
+ manifests, err := snap.Manifests(ctx.fs)
+ if err != nil {
+ return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ entries, err := mf.FetchEntries(ctx.fs, false)
+ if err != nil {
+ return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
+ }
+ for _, e := range entries {
+ if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
+ continue
+ }
+ if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
Review Comment:
Resolved automatically by routing through `validateAddedDataFilesMatchingFilter`. That helper calls `buildPartitionProjection(specID, ...)` keyed per each concurrent manifest's `PartitionSpecID`, so a concurrent file written under spec B is projected against spec B's fields — not spec A's. The eq-delete filter is expressed in row (source) space using `Reference(sourceFieldName)`, so it projects correctly regardless of which spec the concurrent file was written under.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#issuecomment-4386120493
Attached image below
The zizmor failure looks pre-existing. A quick look at the run history:
| Run | PR/Branch | Result |
|-----|-----------|--------|
| #219 | PR #983 (latest push) | ❌ failed |
| #218 | PR #983 | ❌ failed |
| #217 | PR #984 (`feat/geo-type`, different contributor) | ❌ failed |
| #216 | PR #1022 | ✅ passed |
| #215 | PR #983 (earlier today) | ✅ passed |
| #211–#214 | PR #983 | ✅ passed |
PR #983 was passing fine earlier today. PR #984 (completely unrelated, different author) started failing around the same time at run #217. Looks like something changed in the zizmor action around that point, not anything in this diff.
image:
<img width="804" height="597" alt="Screenshot 2026-05-06 at 5 58 48 pm" src="https://github.com/user-attachments/assets/de89eab1-3072-4b81-b411-500f591f5e66" />
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3194249336
##########
table/conflict_validation.go:
##########
@@ -426,6 +427,159 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to the partitions touched
+// by equality-delete files. It builds an OR-of-equalities filter from
+// the provided partition tuples and delegates to
+// validateAddedDataFilesMatchingFilter, which performs per-spec
+// projection, manifest-summary pruning, and type-aware evaluation via
+// iceberg.Literal — making it safe for UUID, decimal, binary, fixed,
+// and future partition types, and correct across partition-spec
+// evolution because each concurrent manifest is projected against its
+// own spec ID.
+//
+// Callers are responsible for ensuring the table is partitioned
+// (i.e. at least one partition field exists) before calling this
+// function. For unpartitioned tables, call
+// validateNoConflictingDataFiles(ctx, iceberg.AlwaysTrue{}, level)
+// directly.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeleteFiles []iceberg.DataFile, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeleteFiles) == 0 {
+ return nil
+ }
+
+ filter, err := eqDeletePartitionsToFilter(eqDeleteFiles, ctx.current)
+ if err != nil {
+ return fmt.Errorf("building partition conflict filter: %w", err)
+ }
+
+ return validateNoConflictingDataFiles(ctx, filter, level)
+}
+
+// eqDeletePartitionsToFilter converts equality-delete data files into an
+// OR-of-ANDs BooleanExpression in row (source) space, suitable for passing
+// to validateAddedDataFilesMatchingFilter.
+//
+// For each eq-delete file it resolves each partition field ID to the source
+// schema field name via the file's partition spec, then builds an EqualTo
+// predicate using Reference(sourceFieldName). Multiple fields within one
+// partition are AND-ed; multiple eq-delete files are OR-ed.
+//
+// The resulting expression is projected per-concurrent-manifest's spec ID
+// inside validateAddedDataFilesMatchingFilter (via buildPartitionProjection),
+// ensuring correct conflict detection even after partition-spec evolution.
+//
+// An empty partition tuple (unpartitioned delete) returns AlwaysTrue so the
+// caller falls back to the conservative full-table scan. Callers should
+// normally guard against calling this function for unpartitioned tables (see
+// RowDelta.validate).
+func eqDeletePartitionsToFilter(files []iceberg.DataFile, meta Metadata) (iceberg.BooleanExpression, error) {
+ terms := make([]iceberg.BooleanExpression, 0, len(files))
+ for _, f := range files {
+ p := f.Partition()
+ if len(p) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ spec := meta.PartitionSpecByID(int(f.SpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("partition spec ID %d not found in metadata", f.SpecID())
+ }
+
+ // Build partition field ID → PartitionField lookup for this spec.
+ partFieldByID := make(map[int]iceberg.PartitionField, spec.NumFields())
+ for _, pf := range spec.Fields() {
+ partFieldByID[pf.FieldID] = pf
+ }
+
+ // Sort partition field IDs for deterministic expression order.
+ fieldIDs := make([]int, 0, len(p))
+ for id := range p {
+ fieldIDs = append(fieldIDs, id)
+ }
+ sort.Ints(fieldIDs)
+
+ conjuncts := make([]iceberg.BooleanExpression, 0, len(p))
+ for _, partFieldID := range fieldIDs {
+ pf, ok := partFieldByID[partFieldID]
+ if !ok {
+ return nil, fmt.Errorf("partition field ID %d not found in spec %d", partFieldID, f.SpecID())
+ }
+
+ // Resolve to source schema field to obtain the Reference name.
+ sourceField, ok := meta.CurrentSchema().FindFieldByID(pf.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("source field ID %d (partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ }
+
+ lit, err := anyToLiteral(p[partFieldID])
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q: %w", sourceField.Name, err)
+ }
+
+ conjuncts = append(conjuncts, iceberg.LiteralPredicate(iceberg.OpEQ, iceberg.Reference(sourceField.Name), lit))
Review Comment:
I think there's a subtle issue here for non-identity transforms — wanted to flag it because the test suite only exercises `IdentityTransform` so it might not surface naturally.
The literal we pass in (`lit`) is the post-transform partition value (a bucket index from `BucketTransform`, days-from-epoch from `DayTransform`, etc.), but the predicate is anchored against the *source* column via `Reference(sourceField.Name)`. Downstream, `validateAddedDataFilesMatchingFilter` calls `BucketTransform.Project` (`transforms.go:366`), which does `transformLiteral(transformer, p.Literal())` — re-bucketing what's already a bucket index.
Quick repro on `BucketTransform{NumBuckets: 16}` over `int64`:
```
user_id=12345 → bucket=1 (what DataFile.Partition() stores)
user_id=1 → bucket=4 (what BucketTransform.Project produces)
```
Eq-delete is in bucket 1, projected predicate matches bucket 4 — a concurrent file in bucket 1 silently passes. Same shape for `day(ts)`, `hour(ts)`, `truncate[K]`, `year`/`month`. `IdentityTransform` happens to work because `identity(identity(x)) == x`, which is why the existing spec-evolution test is green.
A couple of ways to handle, wdyt?
* branch on `pf.Transform`: identity → keep this row-space predicate, non-identity → build a partition-space predicate (`Reference(pf.Name) == lit` keyed by `(specID, pf.Name)`) evaluated only against same-spec concurrent manifests, with cross-spec as a conservative fallback for that file — close to Java's `PartitionSet`.
* narrower scope: explicitly reject non-identity transforms here with a clear error and document this as identity-only for now.
A couple of regression tests under `bucket[N]` and `day(ts)` would lock the contract either way.
##########
table/conflict_validation.go:
##########
@@ -426,6 +427,159 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to the partitions touched
+// by equality-delete files. It builds an OR-of-equalities filter from
+// the provided partition tuples and delegates to
+// validateAddedDataFilesMatchingFilter, which performs per-spec
+// projection, manifest-summary pruning, and type-aware evaluation via
+// iceberg.Literal — making it safe for UUID, decimal, binary, fixed,
+// and future partition types, and correct across partition-spec
+// evolution because each concurrent manifest is projected against its
+// own spec ID.
+//
+// Callers are responsible for ensuring the table is partitioned
+// (i.e. at least one partition field exists) before calling this
+// function. For unpartitioned tables, call
+// validateNoConflictingDataFiles(ctx, iceberg.AlwaysTrue{}, level)
+// directly.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeleteFiles []iceberg.DataFile, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeleteFiles) == 0 {
+ return nil
+ }
+
+ filter, err := eqDeletePartitionsToFilter(eqDeleteFiles, ctx.current)
+ if err != nil {
+ return fmt.Errorf("building partition conflict filter: %w", err)
+ }
+
+ return validateNoConflictingDataFiles(ctx, filter, level)
+}
+
+// eqDeletePartitionsToFilter converts equality-delete data files into an
+// OR-of-ANDs BooleanExpression in row (source) space, suitable for passing
+// to validateAddedDataFilesMatchingFilter.
+//
+// For each eq-delete file it resolves each partition field ID to the source
+// schema field name via the file's partition spec, then builds an EqualTo
+// predicate using Reference(sourceFieldName). Multiple fields within one
+// partition are AND-ed; multiple eq-delete files are OR-ed.
+//
+// The resulting expression is projected per-concurrent-manifest's spec ID
+// inside validateAddedDataFilesMatchingFilter (via buildPartitionProjection),
+// ensuring correct conflict detection even after partition-spec evolution.
+//
+// An empty partition tuple (unpartitioned delete) returns AlwaysTrue so the
+// caller falls back to the conservative full-table scan. Callers should
+// normally guard against calling this function for unpartitioned tables (see
+// RowDelta.validate).
+func eqDeletePartitionsToFilter(files []iceberg.DataFile, meta Metadata) (iceberg.BooleanExpression, error) {
+ terms := make([]iceberg.BooleanExpression, 0, len(files))
+ for _, f := range files {
+ p := f.Partition()
+ if len(p) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ spec := meta.PartitionSpecByID(int(f.SpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("partition spec ID %d not found in metadata", f.SpecID())
+ }
+
+ // Build partition field ID → PartitionField lookup for this spec.
+ partFieldByID := make(map[int]iceberg.PartitionField, spec.NumFields())
+ for _, pf := range spec.Fields() {
+ partFieldByID[pf.FieldID] = pf
+ }
+
+ // Sort partition field IDs for deterministic expression order.
+ fieldIDs := make([]int, 0, len(p))
+ for id := range p {
+ fieldIDs = append(fieldIDs, id)
+ }
+ sort.Ints(fieldIDs)
+
+ conjuncts := make([]iceberg.BooleanExpression, 0, len(p))
+ for _, partFieldID := range fieldIDs {
+ pf, ok := partFieldByID[partFieldID]
+ if !ok {
+ return nil, fmt.Errorf("partition field ID %d not found in spec %d", partFieldID, f.SpecID())
+ }
+
+ // Resolve to source schema field to obtain the Reference name.
+ sourceField, ok := meta.CurrentSchema().FindFieldByID(pf.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("source field ID %d (partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ }
+
+ lit, err := anyToLiteral(p[partFieldID])
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q: %w", sourceField.Name, err)
+ }
+
+ conjuncts = append(conjuncts, iceberg.LiteralPredicate(iceberg.OpEQ, iceberg.Reference(sourceField.Name), lit))
+ }
+
+ if len(conjuncts) == 1 {
+ terms = append(terms, conjuncts[0])
+ } else {
+ terms = append(terms, iceberg.NewAnd(conjuncts[0], conjuncts[1], conjuncts[2:]...))
+ }
+ }
+
+ if len(terms) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ if len(terms) == 1 {
+ return terms[0], nil
+ }
+
+ return iceberg.NewOr(terms[0], terms[1], terms[2:]...), nil
+}
+
+// anyToLiteral converts a dynamically-typed partition value (as
+// stored in iceberg.DataFile.Partition()) to an iceberg.Literal.
+// The supported types mirror the iceberg.LiteralType constraint.
+func anyToLiteral(v any) (iceberg.Literal, error) {
+ switch val := v.(type) {
+ case bool:
+ return iceberg.NewLiteral(val), nil
+ case int32:
+ return iceberg.NewLiteral(val), nil
+ case int64:
+ return iceberg.NewLiteral(val), nil
+ case float32:
+ return iceberg.NewLiteral(val), nil
+ case float64:
+ return iceberg.NewLiteral(val), nil
+ case string:
+ return iceberg.NewLiteral(val), nil
+ case []byte:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Date:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Time:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Timestamp:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.TimestampNano:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Decimal:
Review Comment:
Tiny type-asymmetry I noticed while tracing the round-trip:
`convertAvroValueToIcebergType` at `manifest.go:1800` returns `iceberg.DecimalLiteral{Scale, Val}`, which is `type DecimalLiteral Decimal` — a *named type derived from* `Decimal`, not `Decimal` itself. Go type switches match exact dynamic type, so this arm doesn't fire on a round-tripped value — it falls through to `default:` with `"unsupported partition value type DecimalLiteral"`.
Since `AddDeletes` takes `iceberg.DataFile`, any flow that re-uses a manifest-read file as input would hit this on decimal-partitioned tables. Two options I can think of:
* add `case iceberg.DecimalLiteral:` here
* normalize `convertAvroValueToIcebergType` to return `Decimal` consistently with the writer side
Fwiw `TestAnyToLiteral_SupportedTypes` currently skips Decimal — adding a round-trip case once this is settled would catch any future drift.
##########
table/conflict_validation.go:
##########
@@ -426,6 +427,159 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to the partitions touched
+// by equality-delete files. It builds an OR-of-equalities filter from
+// the provided partition tuples and delegates to
+// validateAddedDataFilesMatchingFilter, which performs per-spec
+// projection, manifest-summary pruning, and type-aware evaluation via
+// iceberg.Literal — making it safe for UUID, decimal, binary, fixed,
+// and future partition types, and correct across partition-spec
+// evolution because each concurrent manifest is projected against its
+// own spec ID.
+//
+// Callers are responsible for ensuring the table is partitioned
+// (i.e. at least one partition field exists) before calling this
+// function. For unpartitioned tables, call
+// validateNoConflictingDataFiles(ctx, iceberg.AlwaysTrue{}, level)
+// directly.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeleteFiles []iceberg.DataFile, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeleteFiles) == 0 {
+ return nil
+ }
+
+ filter, err := eqDeletePartitionsToFilter(eqDeleteFiles, ctx.current)
+ if err != nil {
+ return fmt.Errorf("building partition conflict filter: %w", err)
+ }
+
+ return validateNoConflictingDataFiles(ctx, filter, level)
+}
+
+// eqDeletePartitionsToFilter converts equality-delete data files into an
+// OR-of-ANDs BooleanExpression in row (source) space, suitable for passing
+// to validateAddedDataFilesMatchingFilter.
+//
+// For each eq-delete file it resolves each partition field ID to the source
+// schema field name via the file's partition spec, then builds an EqualTo
+// predicate using Reference(sourceFieldName). Multiple fields within one
+// partition are AND-ed; multiple eq-delete files are OR-ed.
+//
+// The resulting expression is projected per-concurrent-manifest's spec ID
+// inside validateAddedDataFilesMatchingFilter (via buildPartitionProjection),
+// ensuring correct conflict detection even after partition-spec evolution.
+//
+// An empty partition tuple (unpartitioned delete) returns AlwaysTrue so the
+// caller falls back to the conservative full-table scan. Callers should
+// normally guard against calling this function for unpartitioned tables (see
+// RowDelta.validate).
+func eqDeletePartitionsToFilter(files []iceberg.DataFile, meta Metadata) (iceberg.BooleanExpression, error) {
+ terms := make([]iceberg.BooleanExpression, 0, len(files))
+ for _, f := range files {
+ p := f.Partition()
+ if len(p) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ spec := meta.PartitionSpecByID(int(f.SpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("partition spec ID %d not found in metadata", f.SpecID())
+ }
+
+ // Build partition field ID → PartitionField lookup for this spec.
+ partFieldByID := make(map[int]iceberg.PartitionField, spec.NumFields())
+ for _, pf := range spec.Fields() {
+ partFieldByID[pf.FieldID] = pf
+ }
+
+ // Sort partition field IDs for deterministic expression order.
+ fieldIDs := make([]int, 0, len(p))
+ for id := range p {
+ fieldIDs = append(fieldIDs, id)
+ }
+ sort.Ints(fieldIDs)
+
+ conjuncts := make([]iceberg.BooleanExpression, 0, len(p))
+ for _, partFieldID := range fieldIDs {
+ pf, ok := partFieldByID[partFieldID]
+ if !ok {
+ return nil, fmt.Errorf("partition field ID %d not found in spec %d", partFieldID, f.SpecID())
+ }
+
+ // Resolve to source schema field to obtain the Reference name.
+ sourceField, ok := meta.CurrentSchema().FindFieldByID(pf.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("source field ID %d (partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ }
+
+ lit, err := anyToLiteral(p[partFieldID])
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q: %w", sourceField.Name, err)
+ }
+
+ conjuncts = append(conjuncts, iceberg.LiteralPredicate(iceberg.OpEQ, iceberg.Reference(sourceField.Name), lit))
+ }
+
+ if len(conjuncts) == 1 {
+ terms = append(terms, conjuncts[0])
+ } else {
+ terms = append(terms, iceberg.NewAnd(conjuncts[0], conjuncts[1], conjuncts[2:]...))
+ }
+ }
+
+ if len(terms) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ if len(terms) == 1 {
+ return terms[0], nil
+ }
+
+ return iceberg.NewOr(terms[0], terms[1], terms[2:]...), nil
+}
+
+// anyToLiteral converts a dynamically-typed partition value (as
+// stored in iceberg.DataFile.Partition()) to an iceberg.Literal.
+// The supported types mirror the iceberg.LiteralType constraint.
+func anyToLiteral(v any) (iceberg.Literal, error) {
+ switch val := v.(type) {
+ case bool:
+ return iceberg.NewLiteral(val), nil
+ case int32:
+ return iceberg.NewLiteral(val), nil
+ case int64:
+ return iceberg.NewLiteral(val), nil
+ case float32:
+ return iceberg.NewLiteral(val), nil
+ case float64:
+ return iceberg.NewLiteral(val), nil
+ case string:
+ return iceberg.NewLiteral(val), nil
+ case []byte:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Date:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Time:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Timestamp:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.TimestampNano:
Review Comment:
Related to the Decimal one above — `convertAvroValueToIcebergType` (`manifest.go:1756-1817`) has cases for `TimestampMillis` and `TimestampMicros` but none for nanos, so a nanosecond-timestamp partition value arrives here as a raw `int64` and matches `case int64:` first. This `case iceberg.TimestampNano:` arm ends up unreachable from the read path.
Probably worth either adding a `TimestampNanos` case to `convertAvroValueToIcebergType` so reads return `iceberg.TimestampNano` and this arm fires, or dropping this arm and noting that `TimestampNano` partitions aren't supported here yet. Same as Decimal, the unit table also skips this case so it's not caught today.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#issuecomment-4386003693
All points addressed. Summary of what changed:
- **`partitionTupleKey` removed** — replaced by `anyToLiteral` (type switch) + `eqDeletePartitionsToFilter` (builds typed `OR(AND(EqualTo(...)))` filter)
- **Routes through `validateAddedDataFilesMatchingFilter`** — spec evolution, manifest pruning, and type-aware evaluation all handled by the existing path
- **Unpartitioned fallback hoisted to caller** — `RowDelta.validate` checks `NumFields() == 0` and passes `AlwaysTrue` directly
- **Unrelated changes removed** — `catalog/rest/rest.go` and the `newConflictContext` empty-base change are both reverted to match `origin/main`
- **10 tests added** — 6 end-to-end `RowDelta.validate` regression tests (#978 reproducer, same/different partition, UUID type safety, unpartitioned fallback, spec-evolution cross-spec conflict detection) + 4 unit tests for `anyToLiteral` and `validateNoConflictingDataFilesInPartitions` helpers
- **Both `RowDelta` struct doc comment and `validate` function doc updated** to describe the partition-scoped behavior (struct-level comment previously still said "AlwaysTrue")
- **All 19 test packages pass**
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193637795
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
Review Comment:
Moved exactly as suggested. `validateNoConflictingDataFilesInPartitions` no longer contains any `len(p) == 0` fallback. Instead, `RowDelta.validate` checks `currentSpec.NumFields() == 0` before calling anything: if the table is unpartitioned, it calls `validateNoConflictingDataFiles(ctx, iceberg.AlwaysTrue{}, level)` directly; otherwise it calls `validateNoConflictingDataFilesInPartitions`. Both paths are now explicit at the caller.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka merged PR #983:
URL: https://github.com/apache/iceberg-go/pull/983
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193730475
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
Review Comment:
Done — `validateNoConflictingDataFilesInPartitions` is not a sibling that walks manifests independently. It calls `eqDeletePartitionsToFilter` to build an `OR(AND(EqualTo(...)))` expression, then passes it to `validateNoConflictingDataFiles(ctx, filter, level)`, which internally calls `validateAddedDataFilesMatchingFilter`. Per-spec projection, manifest-summary pruning, and type-aware partition evaluation all come for free from the existing path.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3195702570
##########
table/conflict_validation.go:
##########
@@ -426,6 +427,159 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to the partitions touched
+// by equality-delete files. It builds an OR-of-equalities filter from
+// the provided partition tuples and delegates to
+// validateAddedDataFilesMatchingFilter, which performs per-spec
+// projection, manifest-summary pruning, and type-aware evaluation via
+// iceberg.Literal — making it safe for UUID, decimal, binary, fixed,
+// and future partition types, and correct across partition-spec
+// evolution because each concurrent manifest is projected against its
+// own spec ID.
+//
+// Callers are responsible for ensuring the table is partitioned
+// (i.e. at least one partition field exists) before calling this
+// function. For unpartitioned tables, call
+// validateNoConflictingDataFiles(ctx, iceberg.AlwaysTrue{}, level)
+// directly.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeleteFiles []iceberg.DataFile, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeleteFiles) == 0 {
+ return nil
+ }
+
+ filter, err := eqDeletePartitionsToFilter(eqDeleteFiles, ctx.current)
+ if err != nil {
+ return fmt.Errorf("building partition conflict filter: %w", err)
+ }
+
+ return validateNoConflictingDataFiles(ctx, filter, level)
+}
+
+// eqDeletePartitionsToFilter converts equality-delete data files into an
+// OR-of-ANDs BooleanExpression in row (source) space, suitable for passing
+// to validateAddedDataFilesMatchingFilter.
+//
+// For each eq-delete file it resolves each partition field ID to the source
+// schema field name via the file's partition spec, then builds an EqualTo
+// predicate using Reference(sourceFieldName). Multiple fields within one
+// partition are AND-ed; multiple eq-delete files are OR-ed.
+//
+// The resulting expression is projected per-concurrent-manifest's spec ID
+// inside validateAddedDataFilesMatchingFilter (via buildPartitionProjection),
+// ensuring correct conflict detection even after partition-spec evolution.
+//
+// An empty partition tuple (unpartitioned delete) returns AlwaysTrue so the
+// caller falls back to the conservative full-table scan. Callers should
+// normally guard against calling this function for unpartitioned tables (see
+// RowDelta.validate).
+func eqDeletePartitionsToFilter(files []iceberg.DataFile, meta Metadata) (iceberg.BooleanExpression, error) {
+ terms := make([]iceberg.BooleanExpression, 0, len(files))
+ for _, f := range files {
+ p := f.Partition()
+ if len(p) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ spec := meta.PartitionSpecByID(int(f.SpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("partition spec ID %d not found in metadata", f.SpecID())
+ }
+
+ // Build partition field ID → PartitionField lookup for this spec.
+ partFieldByID := make(map[int]iceberg.PartitionField, spec.NumFields())
+ for _, pf := range spec.Fields() {
+ partFieldByID[pf.FieldID] = pf
+ }
+
+ // Sort partition field IDs for deterministic expression order.
+ fieldIDs := make([]int, 0, len(p))
+ for id := range p {
+ fieldIDs = append(fieldIDs, id)
+ }
+ sort.Ints(fieldIDs)
+
+ conjuncts := make([]iceberg.BooleanExpression, 0, len(p))
+ for _, partFieldID := range fieldIDs {
+ pf, ok := partFieldByID[partFieldID]
+ if !ok {
+ return nil, fmt.Errorf("partition field ID %d not found in spec %d", partFieldID, f.SpecID())
+ }
+
+ // Resolve to source schema field to obtain the Reference name.
+ sourceField, ok := meta.CurrentSchema().FindFieldByID(pf.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("source field ID %d (partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ }
+
+ lit, err := anyToLiteral(p[partFieldID])
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q: %w", sourceField.Name, err)
+ }
+
+ conjuncts = append(conjuncts, iceberg.LiteralPredicate(iceberg.OpEQ, iceberg.Reference(sourceField.Name), lit))
+ }
+
+ if len(conjuncts) == 1 {
+ terms = append(terms, conjuncts[0])
+ } else {
+ terms = append(terms, iceberg.NewAnd(conjuncts[0], conjuncts[1], conjuncts[2:]...))
+ }
+ }
+
+ if len(terms) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ if len(terms) == 1 {
+ return terms[0], nil
+ }
+
+ return iceberg.NewOr(terms[0], terms[1], terms[2:]...), nil
+}
+
+// anyToLiteral converts a dynamically-typed partition value (as
+// stored in iceberg.DataFile.Partition()) to an iceberg.Literal.
+// The supported types mirror the iceberg.LiteralType constraint.
+func anyToLiteral(v any) (iceberg.Literal, error) {
+ switch val := v.(type) {
+ case bool:
+ return iceberg.NewLiteral(val), nil
+ case int32:
+ return iceberg.NewLiteral(val), nil
+ case int64:
+ return iceberg.NewLiteral(val), nil
+ case float32:
+ return iceberg.NewLiteral(val), nil
+ case float64:
+ return iceberg.NewLiteral(val), nil
+ case string:
+ return iceberg.NewLiteral(val), nil
+ case []byte:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Date:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Time:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Timestamp:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.TimestampNano:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Decimal:
Review Comment:
Fixed. Added `case iceberg.DecimalLiteral:` to `anyToLiteral` alongside the existing `case iceberg.Decimal:`. The new arm casts to `iceberg.Decimal` before wrapping in a `Literal`, bridging the named-type gap: `iceberg.Decimal(val)` is a valid conversion because `type DecimalLiteral Decimal`.
The `TestAnyToLiteral_SupportedTypes` table now includes a `DecimalLiteral` subtest (`iceberg.DecimalLiteral{Scale: 2}`) that would have failed before this fix — it passes now.
Normalizing `convertAvroValueToIcebergType` to return `Decimal` consistently is deferred; it touches callers beyond `anyToLiteral` and deserves its own PR.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#issuecomment-4388509155
During my integration tests against a real S3-backed Iceberg catalog, I found two pre-existing bugs in `manifest.go` and `literals.go` that surface when any table has a decimal partition column. Neither appears to be tracked in the issue tracker yet, flagging here since they're adjacent to the changes in this PR.
---
**`convertDecimalValue` passes the wrong argument to `DecimalRequiredBytes` (`manifest.go`)**
```go
fixedSize := internal.DecimalRequiredBytes(len(dec.String()))
```
`DecimalRequiredBytes` expects the column's declared **precision** (e.g. `10` for `decimal(10, 2)`), but receives the character length of the printed value instead. These are unrelated: `"10.00"` has length 5, which happens to match `DecimalRequiredBytes(10) = 5` by coincidence, but `"1.00"` has length 4 → `DecimalRequiredBytes(4) = 2` bytes instead of the correct 5. For `decimal(18, 0)`, the value `"1"` gives 1 byte instead of 8.
The resulting slice length doesn't match the Avro schema's `fixed[N]` field, and the encoder rejects it:
```
avro: field data_file.partition.<field>: cannot use []uint8 with Avro type fixed
```
**Repro**: write a manifest with a `decimal(10, 2)` partition column and value `1.00`. The fix is to thread the column's declared precision from the partition field's `DecimalType` into `convertDecimalValue`.
---
**`DecimalLiteral.Type()` hardcodes precision 9 (`literals.go`)**
```go
func (d DecimalLiteral) Type() Type { return DecimalTypeOf(9, d.Scale) }
```
`DecimalLiteral` carries `Val` and `Scale` but not the declared precision, so `Type()` silently returns `decimal(9, scale)` regardless of the actual column type. Any path that calls `.Type()` on a manifest-read decimal literal — bounds checks, cast decisions, predicate projection — will operate against the wrong precision.
---
Both bugs exist in `main` (`v0.5.0`) and are not introduced by this PR. Happy to open separate issues and/or PRs for them if that's preferred.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193679222
##########
table/partition_conflict_test.go:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+package table
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// ---------------------------------------------------------------------------
+// partitionTupleKey
+// ---------------------------------------------------------------------------
+
+func TestPartitionTupleKey_Empty(t *testing.T) {
+ assert.Equal(t, "", partitionTupleKey(map[int]any{}))
+ assert.Equal(t, "", partitionTupleKey(nil))
+}
+
+func TestPartitionTupleKey_SingleField(t *testing.T) {
+ assert.Equal(t, "1=hello;", partitionTupleKey(map[int]any{1: "hello"}))
+}
+
+func TestPartitionTupleKey_MultipleFieldsOrderedByID(t *testing.T) {
+ // Keys must be sorted by field ID, so insertion order must not matter.
+ p1 := map[int]any{1: "a", 3: "c", 2: "b"}
+ p2 := map[int]any{3: "c", 1: "a", 2: "b"}
+ assert.Equal(t, partitionTupleKey(p1), partitionTupleKey(p2),
+ "same content, different insertion order must produce identical keys")
+ assert.Equal(t, "1=a;2=b;3=c;", partitionTupleKey(p1))
+}
+
+func TestPartitionTupleKey_DifferentValuesProduceDifferentKeys(t *testing.T) {
+ pA := map[int]any{1: "us-east-1"}
+ pB := map[int]any{1: "eu-west-1"}
+ assert.NotEqual(t, partitionTupleKey(pA), partitionTupleKey(pB))
+}
+
+// ---------------------------------------------------------------------------
+// validateNoConflictingDataFilesInPartitions — short-circuit paths
+// ---------------------------------------------------------------------------
+
+// TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp
+// verifies that the validator is a no-op under snapshot isolation, matching
+// the behaviour of validateNoConflictingDataFiles.
+func TestValidateNoConflictingDataFilesInPartitions_SnapshotIsolationIsNoOp(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ partitions := []map[int]any{{1: "us-east-1"}}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, partitions, IsolationSnapshot))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_EmptyInputs verifies that
+// the validator short-circuits when there are no concurrent snapshots or no
+// eq-delete partitions, without touching the filesystem.
+func TestValidateNoConflictingDataFilesInPartitions_EmptyInputs(t *testing.T) {
+ head := int64(1)
+ meta := newConflictTestMetadata(t, &head)
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // No concurrent snapshots — must not read any manifests.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{{1: "us-east-1"}}, IsolationSerializable))
+
+ // No eq-delete partitions — nothing to validate.
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, nil, IsolationSerializable))
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{}, IsolationSerializable))
+}
+
+// TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack
+// verifies that a single unpartitioned eq-delete (empty tuple) causes the
+// validator to fall back to the conservative AlwaysTrue path. Because the
+// conflictContext has no concurrent snapshots (fs = nil, concurrent = nil),
+// the AlwaysTrue fallback itself returns nil — this test just ensures we
+// reach and exercise that code path without panicking.
+func TestValidateNoConflictingDataFilesInPartitions_UnpartitionedFallsBack(t *testing.T) {
+ head := int64(1)
+ // Build metadata with a concurrent snapshot by using two different base/current metas.
+ meta := newConflictTestMetadata(t, &head)
+
+ // writerBase has no snapshot; current has head snapshot → head is concurrent.
+ ctx, err := newConflictContext(meta, meta, MainBranch, nil, true)
+ require.NoError(t, err)
+
+ // Empty partition tuple → unpartitioned eq-delete → must fall back to AlwaysTrue.
+ // With no concurrent snapshots in ctx the AlwaysTrue path returns nil (no-op).
+ emptyPartition := map[int]any{}
+ require.NoError(t, validateNoConflictingDataFilesInPartitions(ctx, []map[int]any{emptyPartition}, IsolationSerializable))
Review Comment:
The old trivial-pass test is removed. It is replaced by `TestRowDeltaValidate_UnpartitionedTableFallsBackToAlwaysTrue`, which builds a real table with an empty partition spec, writes a concurrent snapshot with a real data file manifest, and asserts the commit is rejected with `ErrConflictingDataFiles`. This directly proves the `AlwaysTrue` path is active and wired — not just that it compiles.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193668537
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
+ return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
+ }
+ }
+
+ // Build a set of partition tuple keys for O(1) lookup.
+ partSet := make(map[string]struct{}, len(eqDeletePartitions))
+ for _, p := range eqDeletePartitions {
+ partSet[partitionTupleKey(p)] = struct{}{}
+ }
+
+ for _, snap := range ctx.concurrent {
+ manifests, err := snap.Manifests(ctx.fs)
+ if err != nil {
+ return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ entries, err := mf.FetchEntries(ctx.fs, false)
+ if err != nil {
+ return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
+ }
+ for _, e := range entries {
+ if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
+ continue
+ }
+ if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
+ return fmt.Errorf("%w: snapshot %d added data file %s in eq-delete partition",
Review Comment:
Moot once we route through `validateAddedDataFilesMatchingFilter` — that helper produces the richer `"matching filter %s"` message already. No separate fix needed; removing `validateNoConflictingDataFilesInPartitions` eliminates this error site entirely.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3193668537
##########
table/conflict_validation.go:
##########
@@ -426,6 +426,89 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to specific partition
+// tuples derived from equality-delete files. It only flags concurrent
+// data files whose partition tuple matches one of the provided tuples,
+// avoiding false conflicts when a concurrent append lands in a
+// completely different partition.
+//
+// When any provided partition tuple is empty (the table is
+// unpartitioned or the delete file covers all partitions), the check
+// falls back to AlwaysTrue — the equality delete could affect any row.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
+ return nil
+ }
+
+ // If any eq-delete file is unpartitioned (empty tuple), the delete
+ // could affect any row — fall back to the conservative AlwaysTrue check.
+ for _, p := range eqDeletePartitions {
+ if len(p) == 0 {
+ return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
+ }
+ }
+
+ // Build a set of partition tuple keys for O(1) lookup.
+ partSet := make(map[string]struct{}, len(eqDeletePartitions))
+ for _, p := range eqDeletePartitions {
+ partSet[partitionTupleKey(p)] = struct{}{}
+ }
+
+ for _, snap := range ctx.concurrent {
+ manifests, err := snap.Manifests(ctx.fs)
+ if err != nil {
+ return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ entries, err := mf.FetchEntries(ctx.fs, false)
+ if err != nil {
+ return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
+ }
+ for _, e := range entries {
+ if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
+ continue
+ }
+ if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
+ return fmt.Errorf("%w: snapshot %d added data file %s in eq-delete partition",
Review Comment:
Moot — the custom error site no longer exists. By routing through `validateAddedDataFilesMatchingFilter`, the existing error message `"snapshot %d added data file %s matching filter %s"` is produced, which includes the full filter expression. No separate fix needed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "mzzz-zzm (via GitHub)" <gi...@apache.org>.
mzzz-zzm commented on code in PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#discussion_r3195731099
##########
table/conflict_validation.go:
##########
@@ -426,6 +427,159 @@ func validateNoConflictingDataFiles(ctx *conflictContext, filter iceberg.Boolean
return validateAddedDataFilesMatchingFilter(ctx, filter)
}
+// validateNoConflictingDataFilesInPartitions is like
+// validateNoConflictingDataFiles but scoped to the partitions touched
+// by equality-delete files. It builds an OR-of-equalities filter from
+// the provided partition tuples and delegates to
+// validateAddedDataFilesMatchingFilter, which performs per-spec
+// projection, manifest-summary pruning, and type-aware evaluation via
+// iceberg.Literal — making it safe for UUID, decimal, binary, fixed,
+// and future partition types, and correct across partition-spec
+// evolution because each concurrent manifest is projected against its
+// own spec ID.
+//
+// Callers are responsible for ensuring the table is partitioned
+// (i.e. at least one partition field exists) before calling this
+// function. For unpartitioned tables, call
+// validateNoConflictingDataFiles(ctx, iceberg.AlwaysTrue{}, level)
+// directly.
+//
+// Under IsolationSnapshot this validator is a no-op.
+func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeleteFiles []iceberg.DataFile, level IsolationLevel) error {
+ if level != IsolationSerializable {
+ return nil
+ }
+
+ if len(ctx.concurrent) == 0 || len(eqDeleteFiles) == 0 {
+ return nil
+ }
+
+ filter, err := eqDeletePartitionsToFilter(eqDeleteFiles, ctx.current)
+ if err != nil {
+ return fmt.Errorf("building partition conflict filter: %w", err)
+ }
+
+ return validateNoConflictingDataFiles(ctx, filter, level)
+}
+
+// eqDeletePartitionsToFilter converts equality-delete data files into an
+// OR-of-ANDs BooleanExpression in row (source) space, suitable for passing
+// to validateAddedDataFilesMatchingFilter.
+//
+// For each eq-delete file it resolves each partition field ID to the source
+// schema field name via the file's partition spec, then builds an EqualTo
+// predicate using Reference(sourceFieldName). Multiple fields within one
+// partition are AND-ed; multiple eq-delete files are OR-ed.
+//
+// The resulting expression is projected per-concurrent-manifest's spec ID
+// inside validateAddedDataFilesMatchingFilter (via buildPartitionProjection),
+// ensuring correct conflict detection even after partition-spec evolution.
+//
+// An empty partition tuple (unpartitioned delete) returns AlwaysTrue so the
+// caller falls back to the conservative full-table scan. Callers should
+// normally guard against calling this function for unpartitioned tables (see
+// RowDelta.validate).
+func eqDeletePartitionsToFilter(files []iceberg.DataFile, meta Metadata) (iceberg.BooleanExpression, error) {
+ terms := make([]iceberg.BooleanExpression, 0, len(files))
+ for _, f := range files {
+ p := f.Partition()
+ if len(p) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ spec := meta.PartitionSpecByID(int(f.SpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("partition spec ID %d not found in metadata", f.SpecID())
+ }
+
+ // Build partition field ID → PartitionField lookup for this spec.
+ partFieldByID := make(map[int]iceberg.PartitionField, spec.NumFields())
+ for _, pf := range spec.Fields() {
+ partFieldByID[pf.FieldID] = pf
+ }
+
+ // Sort partition field IDs for deterministic expression order.
+ fieldIDs := make([]int, 0, len(p))
+ for id := range p {
+ fieldIDs = append(fieldIDs, id)
+ }
+ sort.Ints(fieldIDs)
+
+ conjuncts := make([]iceberg.BooleanExpression, 0, len(p))
+ for _, partFieldID := range fieldIDs {
+ pf, ok := partFieldByID[partFieldID]
+ if !ok {
+ return nil, fmt.Errorf("partition field ID %d not found in spec %d", partFieldID, f.SpecID())
+ }
+
+ // Resolve to source schema field to obtain the Reference name.
+ sourceField, ok := meta.CurrentSchema().FindFieldByID(pf.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("source field ID %d (partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ }
+
+ lit, err := anyToLiteral(p[partFieldID])
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q: %w", sourceField.Name, err)
+ }
+
+ conjuncts = append(conjuncts, iceberg.LiteralPredicate(iceberg.OpEQ, iceberg.Reference(sourceField.Name), lit))
+ }
+
+ if len(conjuncts) == 1 {
+ terms = append(terms, conjuncts[0])
+ } else {
+ terms = append(terms, iceberg.NewAnd(conjuncts[0], conjuncts[1], conjuncts[2:]...))
+ }
+ }
+
+ if len(terms) == 0 {
+ return iceberg.AlwaysTrue{}, nil
+ }
+
+ if len(terms) == 1 {
+ return terms[0], nil
+ }
+
+ return iceberg.NewOr(terms[0], terms[1], terms[2:]...), nil
+}
+
+// anyToLiteral converts a dynamically-typed partition value (as
+// stored in iceberg.DataFile.Partition()) to an iceberg.Literal.
+// The supported types mirror the iceberg.LiteralType constraint.
+func anyToLiteral(v any) (iceberg.Literal, error) {
+ switch val := v.(type) {
+ case bool:
+ return iceberg.NewLiteral(val), nil
+ case int32:
+ return iceberg.NewLiteral(val), nil
+ case int64:
+ return iceberg.NewLiteral(val), nil
+ case float32:
+ return iceberg.NewLiteral(val), nil
+ case float64:
+ return iceberg.NewLiteral(val), nil
+ case string:
+ return iceberg.NewLiteral(val), nil
+ case []byte:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Date:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Time:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.Timestamp:
+ return iceberg.NewLiteral(val), nil
+ case iceberg.TimestampNano:
Review Comment:
Fixed. Added `case atype.TimestampNanos:` to `convertAvroValueToIcebergType` in `manifest.go`, following the same pattern as the existing `TimestampMillis` and `TimestampMicros` cases. A `timestamp-nanos` logical type field now returns `iceberg.TimestampNano` from the manifest reader, making the `anyToLiteral` arm reachable from the read path.
Added a `TimestampNano` subtest to `TestAnyToLiteral_SupportedTypes` to catch future drift. The test passes.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(occ): RowDelta uses partition-scoped conflict check instead of AlwaysTrue [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on PR #983:
URL: https://github.com/apache/iceberg-go/pull/983#issuecomment-4388700435
@mzzz-zzm --> for those new bugs, can you file a separate issues? Thanks for this! It's very usefull
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org