zeroshade commented on code in PR #1607: URL: https://github.com/apache/iceberg-go/pull/1607#discussion_r3732110360
########## variant_cast.go: ########## @@ -0,0 +1,269 @@ +// 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 iceberg + +import ( + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/decimal" + "github.com/apache/arrow-go/v18/arrow/decimal128" + "github.com/apache/arrow-go/v18/parquet/variant" + "github.com/google/uuid" +) + +const ( + microsPerDay = int64(86_400_000_000) + nanosPerDay = int64(86_400_000_000_000) + nanosPerMicro = int64(1_000) +) + +// CastVariantLiteral casts a leaf variant value to typ and wraps it as a Literal. +func CastVariantLiteral(v variant.Value, typ PrimitiveType) (Literal, bool) { + result, ok := castVariantValue(v, typ) + if !ok { + return nil, false + } + + lit := literalFromCastValue(result) + if lit == nil { + return nil, false + } + if !lit.Type().Equals(typ) { + conv, err := lit.To(typ) + if err != nil { + return nil, false + } + + lit = conv + } + + return lit, true +} + +// castVariantValue casts a leaf variant value to the Go value backing typ. +func castVariantValue(v variant.Value, typ PrimitiveType) (any, bool) { + raw := v.Value() + if raw == nil { + return nil, false + } + + if r, ok := exactVariantMatch(v.Type(), raw, typ); ok { + return r, true + } + + switch t := typ.(type) { + case Int32Type: + switch n := raw.(type) { + case int8: + return int32(n), true + case int16: + return int32(n), true + } + case Int64Type: + switch n := raw.(type) { + case int8: + return int64(n), true + case int16: + return int64(n), true + case int32: + return int64(n), true + } + case Float64Type: + if f, ok := raw.(float32); ok { + return float64(f), true + } + case FixedType: + if b, ok := raw.([]byte); ok && len(b) == t.Len() { + return b, true + } + case DecimalType: + return castVariantDecimal(raw, t) + case BooleanType: + if b, ok := raw.(bool); ok { + return b, true + } + case TimestampType, TimestampTzType: Review Comment: **Blocking: this conflates tz-aware and zoneless timestamps, and it produces wrong rows rather than just wrong pruning.** `exactVariantMatch` (`variant_cast.go:62-84`) gets this right — `TimestampMicrosNTZ` -> `timestamp`, `TimestampMicros` -> `timestamptz`. But the fallback dispatch here drops the distinction entirely: both `TimestampType` and `TimestampTzType` route to the same `castVariantToMicros`, and both `TimestampNsType` and `TimestampTzNsType` route to the same `castVariantToNanos`. `castVariantToMicros` (`:195-227`) then accepts both `variant.TimestampNanos` *and* `variant.TimestampNanosNTZ` regardless of which target was requested. Net effect: `Extract(p, "$.ts", TimestampTz)` silently matches a zoneless nanosecond leaf, treating local time as UTC, and `Extract(p, "$.ts", Timestamp)` silently matches a tz-aware nanosecond leaf. Since this feeds both file pruning and the residual filter, the rows that come back are wrong. The inconsistency with `exactVariantMatch` reads as an oversight rather than a decision. Suggested fix: pass the target's tz-awareness into the three helpers and require the source physical type to agree. ```go func castVariantToMicros(pt variant.Type, raw any, tz bool) (any, bool) { switch { case tz && pt == variant.TimestampNanos, !tz && pt == variant.TimestampNanosNTZ: return Timestamp(floorDiv(int64(raw.(arrow.Timestamp)), nanosPerMicro)), true ... ``` ########## table/variant_residual.go: ########## @@ -0,0 +1,193 @@ +// 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 ( + "context" + "fmt" + "strconv" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/extensions" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// augmentSchemaWithExtracts returns fileSchema plus one primitive column per variant extract term. +func augmentSchemaWithExtracts(fileSchema *iceberg.Schema, cols []iceberg.VariantExtractColumn) (*iceberg.Schema, error) { + fields := fileSchema.Fields() + for _, c := range cols { + fields = append(fields, iceberg.NestedField{ + ID: c.FieldID, + Name: c.Name, + Type: c.Term.Type().(iceberg.PrimitiveType), + }) + } + + return iceberg.NewSchema(fileSchema.ID, fields...), nil +} + +// buildExtractColumn materializes one variant extract term into a typed Arrow array over rec. +func buildExtractColumn(col iceberg.VariantExtractColumn, rec arrow.RecordBatch, mem memory.Allocator) (arrow.Array, arrow.Field, error) { + typ := col.Term.Type().(iceberg.PrimitiveType) + dt, err := TypeToArrowType(typ, false, false) + if err != nil { + return nil, arrow.Field{}, err + } + + bldr := array.NewBuilder(mem, dt) + defer bldr.Release() + + n := int(rec.NumRows()) + varIdx := fieldIndexByID(rec.Schema(), col.Term.Ref().Field().ID) + varr, _ := columnAt(rec, varIdx).(*extensions.VariantArray) Review Comment: **Blocking: the discarded `ok` turns a lookup miss into a silently emptied batch.** If `fieldIndexByID` misses (returns -1) or the column comes back as something other than a `*extensions.VariantArray`, `varr` is nil, the loop at `:72` appends null for *every* row, and `payload.a = 5` then evaluates false everywhere. That is a wrong-results failure wearing the costume of an empty result — nothing errors and nothing logs. `varIdx == -1` in particular is reachable when the arrow field lacks `PARQUET:field_id` metadata (name-mapping reads), which is precisely the schema-evolution case where "no rows" is indistinguishable from a legitimate answer. Suggested fix: distinguish the two cases. Return an explicit error when the column is present but is not a `VariantArray`, and handle "column absent from this file" the way `scanTranslator.VisitBound` already does for a missing column — `AlwaysFalse` for comparison ops, `AlwaysTrue` for `IsNull` — rather than fabricating an all-null column. ########## variant_extract.go: ########## @@ -0,0 +1,248 @@ +// 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 iceberg + +import ( + "fmt" + + "github.com/apache/arrow-go/v18/parquet/variant" + "github.com/google/uuid" +) + +// BoundExtract is a bound variant sub-path term used for metrics pruning and residual evaluation. +type BoundExtract interface { + BoundTerm + + Path() string + // ExtractValue navigates v to this term's path and casts the leaf to the target type. + ExtractValue(v variant.Value) (Literal, bool) +} + +// Extract creates an unbound variant sub-path term for a dotted JSONPath. +func Extract(ref Reference, path string, typ PrimitiveType) UnboundTerm { + return &unboundExtract{ref: ref, path: path, typ: typ} +} + +type unboundExtract struct { Review Comment: **Blocking: extract terms serialize to `"term":{}` with no error, corrupting REST scan-planning filters.** Neither `unboundExtract` nor `boundExtract[T]` has a `MarshalJSON`. `marshalLiteralPredicate` (`expr_json.go:230-239`) calls `json.Marshal(term)`, and a struct with only unexported fields marshals to `{}`. So a row filter containing an extract, sent through server-side scan planning (`catalog/rest/scan_planning.go:389`), is encoded as `{"type":"eq","term":{},"value":5}` and shipped to the catalog with no client-side error. There is also no `decodeTerm` case, so it cannot round-trip back. The repo already has the right precedent for a non-serializable term — `expr_json.go:204-205` returns `ErrBBoxNotSerializable`. Suggested fix: ```go func (u *unboundExtract) MarshalJSON() ([]byte, error) { return nil, fmt.Errorf("%w: variant extract terms have no REST expression form", ErrNotImplemented) } func (b *boundExtract[T]) MarshalJSON() ([]byte, error) { /* same */ } ``` ########## variant_path.go: ########## @@ -0,0 +1,132 @@ +// 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 iceberg + +import ( + "fmt" + "strings" +) + +// NormalizeVariantPath renders member names as the spec's RFC-9535 normalized JSON path. +func NormalizeVariantPath(fields []string) string { + if len(fields) == 0 { + return "$" + } + + var b strings.Builder + b.WriteByte('$') + for _, f := range fields { + b.WriteString("['") + b.WriteString(rfc9535Escape(f)) + b.WriteString("']") + } + + return b.String() +} + +func rfc9535Escape(name string) string { + if strings.IndexFunc(name, func(r rune) bool { + return r < 0x20 || r == '\'' || r == '\\' + }) < 0 { + return name + } + + var b strings.Builder + b.Grow(len(name) + 4) + for _, r := range name { + switch r { + case '\b': + b.WriteString(`\b`) + case '\t': + b.WriteString(`\t`) + case '\f': + b.WriteString(`\f`) + case '\n': + b.WriteString(`\n`) + case '\r': + b.WriteString(`\r`) + case '\'': + b.WriteString(`\'`) + case '\\': + b.WriteString(`\\`) + default: + if r < 0x20 { + fmt.Fprintf(&b, `\u%04x`, r) + } else { + b.WriteRune(r) + } + } + } + + return b.String() +} + +// parseVariantPath parses a dot-shorthand variant path ($.a.b) into its member names. +func parseVariantPath(path string) ([]string, error) { Review Comment: Non-blocking round-trip asymmetry: `parseVariantPath` rejects bracket notation, but `BoundExtract.Path()` (`variant_extract.go:146`) *emits* it. So `Extract(ref, term.Path(), typ)` cannot round-trip a path that normalizes to brackets. It also makes field names containing `.`, quotes, or a leading digit unreachable, which Java's `PathUtil.parse` supports. Suggested fix: accept `$['a']` on input even if array indices stay unsupported for now. ########## table/evaluators.go: ########## @@ -868,25 +880,44 @@ func (m *inclusiveMetricsEval) VisitNotNan(t iceberg.BoundTerm) bool { return rowsMightMatch } -func (m *inclusiveMetricsEval) VisitLess(t iceberg.BoundTerm, lit iceberg.Literal) bool { - field := t.Ref().Field() - fieldID := field.ID +// boundFor decodes the file bound for term t from raw: a scalar for a reference, or the Review Comment: Non-blocking, just flagging an unremarked behavior change: the non-primitive `panic` guard in `boundFor` now fires only when `raw != nil`, whereas previously it fired unconditionally in `VisitLess`, `VisitIn`, `VisitStartsWith`, and friends. That is more permissive and is very likely fine, but it is not mentioned anywhere in the PR description. Suggested fix: confirm the loosening is intended and note it in the description (or in a comment here). ########## table/evaluators.go: ########## @@ -868,25 +880,44 @@ func (m *inclusiveMetricsEval) VisitNotNan(t iceberg.BoundTerm) bool { return rowsMightMatch } -func (m *inclusiveMetricsEval) VisitLess(t iceberg.BoundTerm, lit iceberg.Literal) bool { - field := t.Ref().Field() - fieldID := field.ID +// boundFor decodes the file bound for term t from raw: a scalar for a reference, or the +// variant sub-path value for an extract; ok is false when raw is nil or not castable. +func (m *inclusiveMetricsEval) boundFor(t iceberg.BoundTerm, raw []byte) (iceberg.Literal, bool) { + if raw == nil { + return nil, false + } - if m.containsNullsOnly(fieldID) || m.containsNansOnly(fieldID) { - return rowsCannotMatch + if ext, ok := t.(iceberg.BoundExtract); ok { + lit, found, err := internal.VariantBoundLiteral(raw, ext.Path(), ext.Type().(iceberg.PrimitiveType)) + if err != nil { + panic(err) Review Comment: Non-blocking: `panic(err)` turns an undecodable bound into a hard scan error — for example a bound written by another engine, or a `TestRowGroup` path where `lowerBounds[-1]` holds arbitrary variant-leaf stats. Metrics pruning should be best-effort; a bound we cannot decode means we cannot prune, not that the scan should fail. Worth noting the round trip is self-defeating as written: `VariantBoundLiteral` installs a `recover()` (`table/internal/variant_bounds.go:317-321`) specifically to convert arrow-go panics into errors, and `boundFor` immediately re-panics them. Suggested fix: return `rowsMightMatch` here instead of panicking. ########## table/variant_residual.go: ########## @@ -0,0 +1,193 @@ +// 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 ( + "context" + "fmt" + "strconv" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/extensions" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// augmentSchemaWithExtracts returns fileSchema plus one primitive column per variant extract term. +func augmentSchemaWithExtracts(fileSchema *iceberg.Schema, cols []iceberg.VariantExtractColumn) (*iceberg.Schema, error) { + fields := fileSchema.Fields() + for _, c := range cols { + fields = append(fields, iceberg.NestedField{ + ID: c.FieldID, + Name: c.Name, + Type: c.Term.Type().(iceberg.PrimitiveType), + }) + } + + return iceberg.NewSchema(fileSchema.ID, fields...), nil +} + +// buildExtractColumn materializes one variant extract term into a typed Arrow array over rec. +func buildExtractColumn(col iceberg.VariantExtractColumn, rec arrow.RecordBatch, mem memory.Allocator) (arrow.Array, arrow.Field, error) { + typ := col.Term.Type().(iceberg.PrimitiveType) + dt, err := TypeToArrowType(typ, false, false) + if err != nil { + return nil, arrow.Field{}, err + } + + bldr := array.NewBuilder(mem, dt) + defer bldr.Release() + + n := int(rec.NumRows()) + varIdx := fieldIndexByID(rec.Schema(), col.Term.Ref().Field().ID) + varr, _ := columnAt(rec, varIdx).(*extensions.VariantArray) + + for i := 0; i < n; i++ { + if varr == nil || varr.IsNull(i) { + bldr.AppendNull() + + continue + } + + v, verr := varr.Value(i) + if verr != nil { + bldr.AppendNull() Review Comment: Non-blocking: a variant decode failure (`verr != nil`) is silently converted into "no match" for that row. That is a different thing from the value being absent, and the caller has no way to tell them apart. Suggested fix: at minimum log it; ideally surface it, since a decode failure usually means the file is malformed rather than that the path is missing. ########## table/variant_residual.go: ########## @@ -0,0 +1,193 @@ +// 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 ( + "context" + "fmt" + "strconv" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/extensions" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// augmentSchemaWithExtracts returns fileSchema plus one primitive column per variant extract term. +func augmentSchemaWithExtracts(fileSchema *iceberg.Schema, cols []iceberg.VariantExtractColumn) (*iceberg.Schema, error) { Review Comment: Minor: `augmentSchemaWithExtracts` declares an `error` return it never produces — every caller has to handle an impossible case. Suggested fix: drop the return value. Separately, `iceberg.NewSchema(fileSchema.ID, fields...)` on `:45` drops the source schema's identifier field IDs; harmless for binding today, but easy to preserve while you are here. ########## variant_path.go: ########## @@ -0,0 +1,132 @@ +// 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 iceberg + +import ( + "fmt" + "strings" +) + +// NormalizeVariantPath renders member names as the spec's RFC-9535 normalized JSON path. +func NormalizeVariantPath(fields []string) string { Review Comment: Minor API-surface note: `NormalizeVariantPath` and `CastVariantLiteral` (`variant_cast.go:41`) appear to be exported only so `table/internal` can reach them. That permanently widens the root package's public surface to serve an internal need. Suggested fix: consider moving `VariantBoundLiteral`'s decode into the root package instead, or at minimum document both as internal-use and not covered by compatibility guarantees. -- 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]
