DmitryKotuseu opened a new issue, #1876:
URL: https://github.com/apache/iceberg-go/issues/1876
### 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:
```go
// 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 — v0.6.0 panics earlier when reading any
decimal column (`interface conversion: iceberg.Type is iceberg.DecimalType,
not
*iceberg.DecimalType`), which appears to be already fixed on main.
**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 #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:
```go
// 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)
}
})
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
