Row filter on an INT32/INT64-backed decimal column silently matches zero rows: Parquet row-group statistics are decoded as Iceberg bounds

[DmitryKotuseu]

Apache Iceberg version

main (development)

Please describe the bug 🐞

A row filter on a decimal column whose Parquet physical type is INT32 or INT64
silently matches zero rows. No error is returned. An unfiltered read of the
same column returns every row with correct values.

Root cause. The metrics evaluator's bounds maps hold Iceberg single-value
encodings — they are consumed with iceberg.LiteralFromBytes(field.Type, bytes)
(table/evaluators.go:900, :932, :1027). The manifest path (Eval) fills
them correctly from Iceberg metadata. The row-group path does not:

// table/evaluators.go:790-792, inclusiveMetricsEval.TestRowGroup
if stats.HasMinMax() {
    m.lowerBounds[fieldID] = stats.EncodeMin()   // Parquet plain encoding
    m.upperBounds[fieldID] = stats.EncodeMax()
}

EncodeMin/EncodeMax return Parquet's plain encoding, which for an
INT32/INT64-backed column is little-endian. Iceberg serialises a decimal as
minimum-length big-endian two's complement. Decimal is the only type where
the two disagree: int/long/float/double/date are little-endian on both sides,
string/binary are raw bytes on both, and a FIXED_LEN_BYTE_ARRAY-backed decimal
is big-endian two's complement in Parquet's plain encoding too — which is
exactly why this has gone unnoticed.

Worked example, unscaled value 659 in a decimal(9,0) column:

parquet stats bytes         93 02 00 00     (little-endian int32)
decoded as an Iceberg bound 0x93020000  =  -1828585472   (leading bit set -> negative)

Every row group then reports bounds [-1828585472, -1828585472], so
VisitEqual and VisitGreaterThanEqual return rowsCannotMatch for every row
group of every file and the scan yields nothing. LessThanEqual survives
because a hugely negative lower bound still admits "might match", which makes
the failure look erratic:

predicate over 4096 rows all holding 659	matched
no filter	4096 (correct)
LessThanEqual(659)	4096 (correct)
EqualTo(659)	0
IsIn(659)	0
GreaterThanEqual(659)	0

This fires on spec-conformant files. The Iceberg spec's Parquet mapping
prescribes INT32 for decimal precision ≤ 9 and INT64 for ≤ 18, and iceberg-go's
own writer emits those types for shredded decimals
(table/internal/parquet_files.go:699-701, "Shredded decimals need the spec's
INT32/INT64/FLBA-by-precision types"). So the library can silently mis-prune
files it wrote itself. It is not limited to foreign writers, though that is how
we hit it: reading Snowflake Open Catalog, where a decimal(38,0) column is
stored as physical=INT32 logical=Decimal(precision=3, scale=0), a scan of
172,417 rows that all match returned 0 rows across 16 planned files, with no
error.

Reproduction. Attached decimal_rowgroup_stats_test.go — self-contained, no
credentials, no network. It writes one Parquet file per case (via
WithStoreDecimalAsInteger for the INT32 case), registers it with AddFiles
on a hadoop catalog, and filters:

INT32-backed decimal(9,0)                     unfiltered=4096 EqualTo=0    GreaterThanEqual=0    LessThanEqual=4096   FAIL
FIXED_LEN_BYTE_ARRAY decimal(38,0) [control]  unfiltered=4096 EqualTo=4096 GreaterThanEqual=4096 LessThanEqual=4096   PASS

The same construct is present in v0.6.0 at table/evaluators.go:755, though I have not run the reproduction there: on v0.6.0 this fixture fails earlier, in the AddFiles path, with interface conversion: iceberg.Type is iceberg.DecimalType, not *iceberg.DecimalType. That path works on main. Reading decimal columns from tables written by iceberg-go itself works on v0.6.0 too, so the v0.6.0 behaviour of the bug reported here is untested rather than known-different.

Suggested fix. Two parts:

1. Correctness first. When the Parquet physical type is INT32/INT64 and the
   Iceberg field is a decimal, leave that column out of the bounds maps so the
   evaluator falls back to rowsMightMatch. This mirrors the remedy already
   accepted in fix(table): guard malformed parquet UUID stats #1414 for malformed UUID stats, and the existing
   ErrInvalidFixedLength -> rowsMightMatch fallback in TestRowGroup.
2. Then restore pruning. Build the bound from the typed statistic —
   stats.Min()/stats.Max() give int32/int64 — as
   iceberg.DecimalLiteral{Val: decimal128.FromI64(v), Scale: s}.MarshalBinary(),
   taking the scale from the Parquet logical type and rescaling to the Iceberg
   field's scale when they differ (writers may narrow it, as the Snowflake case
   above narrows precision from 38 to 3).

Test:

// Self-contained reproduction: a row filter on a decimal column backed by an
// INT32/INT64 Parquet physical type silently matches zero rows.
//
// No credentials, no network. Run with:  go test -v ./...
package repro

import (
	"context"
	"os"
	"path/filepath"
	"testing"

	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/decimal128"
	"github.com/apache/arrow-go/v18/arrow/memory"
	"github.com/apache/arrow-go/v18/parquet"
	"github.com/apache/arrow-go/v18/parquet/pqarrow"
	"github.com/apache/iceberg-go"
	"github.com/apache/iceberg-go/catalog"
	"github.com/apache/iceberg-go/catalog/hadoop"
	"github.com/apache/iceberg-go/table"
)

const (
	colName = "ORG_ID"
	value   = 659
	rows    = 4096
)

// writeParquet writes one file whose single decimal column uses the physical
// type the Iceberg spec prescribes for its precision: INT32 for precision <= 9,
// FIXED_LEN_BYTE_ARRAY for precision 38. arrow-go emits the former only with
// WithStoreDecimalAsInteger, which is exactly what iceberg-go's own writer
// enables for shredded decimals (table/internal/parquet_files.go).
func writeParquet(t *testing.T, dir string, precision int, asInteger bool) string {
	t.Helper()

	dt := &arrow.Decimal128Type{Precision: int32(precision), Scale: 0}
	schema := arrow.NewSchema([]arrow.Field{{
		Name: colName, Type: dt, Nullable: true,
		Metadata: arrow.NewMetadata([]string{"PARQUET:field_id"}, []string{"1"}),
	}}, nil)

	bldr := array.NewDecimal128Builder(memory.DefaultAllocator, dt)
	defer bldr.Release()
	for i := 0; i < rows; i++ {
		bldr.Append(decimal128.FromI64(value))
	}
	col := bldr.NewArray()
	defer col.Release()

	rec := array.NewRecordBatch(schema, []arrow.Array{col}, rows)
	defer rec.Release()
	tbl := array.NewTableFromRecords(schema, []arrow.RecordBatch{rec})
	defer tbl.Release()

	path := filepath.Join(dir, "data.parquet")
	f, err := os.Create(path)
	if err != nil {
		t.Fatalf("create: %v", err)
	}
	// WriteTable closes f.
	if err := pqarrow.WriteTable(tbl, f, rows,
		parquet.NewWriterProperties(
			parquet.WithStoreDecimalAsInteger(asInteger),
			parquet.WithStats(true)),
		pqarrow.NewArrowWriterProperties(pqarrow.WithStoreSchema())); err != nil {
		t.Fatalf("WriteTable: %v", err)
	}

	return path
}

func newTable(t *testing.T, precision int, asInteger bool) *table.Table {
	t.Helper()

	root := t.TempDir()
	data := filepath.Join(root, "data")
	if err := os.MkdirAll(data, 0o755); err != nil {
		t.Fatalf("mkdir: %v", err)
	}
	path := writeParquet(t, data, precision, asInteger)

	warehouse := filepath.Join(root, "warehouse")
	if err := os.MkdirAll(warehouse, 0o755); err != nil {
		t.Fatalf("mkdir warehouse: %v", err)
	}

	cat, err := hadoop.NewCatalog("local", warehouse, iceberg.Properties{})
	if err != nil {
		t.Fatalf("NewCatalog: %v", err)
	}

	ctx := context.Background()
	if err := cat.CreateNamespace(ctx, table.Identifier{"ns"}, nil); err != nil {
		t.Fatalf("CreateNamespace: %v", err)
	}

	tbl, err := cat.CreateTable(ctx, table.Identifier{"ns", "t"},
		iceberg.NewSchema(0, iceberg.NestedField{
			ID: 1, Name: colName, Type: iceberg.DecimalTypeOf(precision, 0), Required: false,
		}),
		catalog.WithProperties(iceberg.Properties{"format-version": "2"}))
	if err != nil {
		t.Fatalf("CreateTable: %v", err)
	}

	txn := tbl.NewTransaction()
	if err := txn.AddFiles(ctx, []string{path}, nil, false); err != nil {
		t.Fatalf("AddFiles: %v", err)
	}
	added, err := txn.Commit(ctx)
	if err != nil {
		t.Fatalf("Commit: %v", err)
	}

	return added
}

func count(t *testing.T, scan *table.Scan) int64 {
	t.Helper()

	_, batches, err := scan.ToArrowRecords(context.Background())
	if err != nil {
		t.Fatalf("ToArrowRecords: %v", err)
	}

	var n int64
	for b, err := range batches {
		if err != nil {
			t.Fatalf("batch: %v", err)
		}
		n += b.NumRows()
		b.Release()
	}

	return n
}

// TestDecimalRowFilterOnIntBackedColumn: every row holds 659, so every
// predicate below must match all 4096 rows. With an INT32-backed decimal,
// EqualTo and GreaterThanEqual match none and no error is returned.
func TestDecimalRowFilterOnIntBackedColumn(t *testing.T) {
	for _, tc := range []struct {
		name      string
		precision int
		asInteger bool
	}{
		{"INT32-backed decimal(9,0)", 9, true},
		{"FIXED_LEN_BYTE_ARRAY decimal(38,0) [control]", 38, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			tbl := newTable(t, tc.precision, tc.asInteger)

			ref := iceberg.Reference(colName)
			lit := iceberg.Decimal{Val: decimal128.FromI64(value), Scale: 0}

			unfiltered := count(t, tbl.Scan(table.WithSelectedFields(colName)))
			eq := count(t, tbl.Scan(table.WithRowFilter(iceberg.EqualTo(ref, lit))))
			ge := count(t, tbl.Scan(table.WithRowFilter(iceberg.GreaterThanEqual(ref, lit))))
			le := count(t, tbl.Scan(table.WithRowFilter(iceberg.LessThanEqual(ref, lit))))

			t.Logf("unfiltered=%d EqualTo=%d GreaterThanEqual=%d LessThanEqual=%d",
				unfiltered, eq, ge, le)

			if unfiltered != rows {
				t.Fatalf("unfiltered read %d rows, want %d", unfiltered, rows)
			}
			if eq != rows {
				t.Errorf("EqualTo matched %d rows, want %d", eq, rows)
			}
			if ge != rows {
				t.Errorf("GreaterThanEqual matched %d rows, want %d", ge, rows)
			}
			if le != rows {
				t.Errorf("LessThanEqual matched %d rows, want %d", le, rows)
			}
		})
	}
}

[Revanth14]

Thanks for the detailed report and reproducer. @DmitryKotuseu
Are you planning to open a PR for this? If not, I’d be happy to work on the fix.

[DmitryKotuseu]

@Revanth14. Please go ahead - I won't have time to take it, and I'd rather not hold it up.
Two things that might help:
The reproducer covers the spec-conformant case, where the Iceberg field is decimal(9,0) and the file stores it as INT32 with Decimal(9,0). The case we actually hit in the wild narrows the logical type as well: the Iceberg schema says decimal(38,0) and Snowflake writes physical=INT32 logical=Decimal(precision=3, scale=0). Worth a second test case, since that's the shape a real writer produces, and it's the one where the scale has to be taken from the Parquet logical type rather than assumed to match the field.
If part 1 of the suggested fix (drop the bounds and fall back to rowsMightMatch) is easier to land on its own, that alone restores correctness - pruning can follow separately.
Happy to test a PR against the live Snowflake Open Catalog table where we found this, which is the one case that can't be reproduced locally. Just ping me.