zeroshade commented on code in PR #2395:
URL: https://github.com/apache/iceberg-go/pull/2395#discussion_r4469986212
##########
table/arrow_utils.go:
##########
@@ -1945,29 +2133,51 @@ func fileToDataFile(ctx context.Context, fileIO
iceio.IO, filePath string, curre
arrSchema,
)
- partitionValues := make(map[int]any)
- if !currentSpec.Equals(*iceberg.UnpartitionedSpec) {
- for _, field := range currentSpec.Fields() {
- if !field.Transform.PreservesOrder() {
- panic(fmt.Errorf("cannot infer partition value
from parquet metadata for a non-linear partition field: %s with transform %s",
field.Name, field.Transform))
- }
+ // For bucket specs, require the complete tuple instead of combining
caller
+ // values with footer inference, which cannot prove null partition
values.
+ hasBucket := false
+ for _, field := range currentSpec.Fields() {
+ switch field.Transform.(type) {
+ case iceberg.BucketTransform, *iceberg.BucketTransform:
Review Comment:
Supplied values skip the bounds check that inferred values get. For
order-preserving transforms, please compare the supplied value with the result
of applying the transform to the file's lower and upper bounds, as the
inference path does.
##########
table/arrow_utils.go:
##########
@@ -1897,6 +1901,166 @@ func filesToDataFiles(ctx context.Context, fileIO
iceio.IO, meta *MetadataBuilde
return dataFiles, nil
}
+// FileToDataFileOption configures the conversion of an existing Parquet file.
+type FileToDataFileOption func(*fileToDataFileConfig)
+
+type fileToDataFileConfig struct {
+ partitionValues map[int]any
+}
+
+// WithFilePartitionValues supplies the partition tuple of an existing data
file,
+// keyed by partition field ID (not source column ID). This is required for
+// transforms such as bucket[N], which cannot be inferred from Parquet
statistics.
+// For a spec containing a bucket transform, all non-void fields must be
supplied.
+// Each supplied value must match the transform's result type. A nil value
+// explicitly represents a null partition. Callers must ensure that every row
+// in the file has the supplied partition value, as required by the Iceberg
spec.
+func WithFilePartitionValues(values map[int]any) FileToDataFileOption {
+ partitionValues := cloneFilePartitionTuple(values)
+
+ return func(cfg *fileToDataFileConfig) {
+ cfg.partitionValues = partitionValues
+ }
+}
+
+// cloneFilePartitionTuple prevents the caller from changing partition values
+// while concurrently imported Parquet files are being inspected.
+func cloneFilePartitionTuple(values map[int]any) map[int]any {
+ if values == nil {
+ return nil
+ }
+ cloned := make(map[int]any, len(values))
+ for id, value := range values {
+ if raw, ok := value.([]byte); ok {
+ value = bytes.Clone(raw)
+ }
+ cloned[id] = value
+ }
+
+ return cloned
+}
+
+func validateFileBucketValue(field iceberg.PartitionField, value any,
numBuckets int) error {
+ if value == nil {
+ return nil
+ }
+
+ bucket, ok := value.(int32)
+ if !ok || bucket < 0 || int64(bucket) >= int64(numBuckets) {
+ return fmt.Errorf("%w: partition field %q (id %d) requires an
int32 bucket in [0, %d), got %v (%T)",
+ iceberg.ErrInvalidArgument, field.Name, field.FieldID,
numBuckets, value, value)
+ }
+
+ return nil
+}
+
+func validFilePartitionValueType(typ iceberg.Type, value any) bool {
+ switch t := typ.(type) {
+ case iceberg.BooleanType:
+ _, ok := value.(bool)
+
+ return ok
+ case iceberg.Int32Type:
+ _, ok := value.(int32)
+
+ return ok
+ case iceberg.Int64Type:
+ _, ok := value.(int64)
+
+ return ok
+ case iceberg.Float32Type:
+ _, ok := value.(float32)
+
+ return ok
+ case iceberg.Float64Type:
+ _, ok := value.(float64)
+
+ return ok
+ case iceberg.StringType:
+ _, ok := value.(string)
+
+ return ok
+ case iceberg.BinaryType:
+ _, ok := value.([]byte)
+
+ return ok
+ case iceberg.FixedType:
+ v, ok := value.([]byte)
+
+ return ok && len(v) == t.Len()
+ case iceberg.DateType:
+ _, ok := value.(iceberg.Date)
+
+ return ok
+ case iceberg.TimeType:
+ _, ok := value.(iceberg.Time)
+
+ return ok
+ case iceberg.TimestampType, iceberg.TimestampTzType:
+ _, ok := value.(iceberg.Timestamp)
+
+ return ok
+ case iceberg.TimestampNsType, iceberg.TimestampTzNsType:
+ _, ok := value.(iceberg.TimestampNano)
+
+ return ok
+ case iceberg.UUIDType:
+ _, ok := value.(uuid.UUID)
+
+ return ok
+ case iceberg.DecimalType:
+ v, ok := value.(iceberg.Decimal)
+
+ return ok && v.Scale == t.Scale() &&
v.Val.FitsInPrecision(int32(t.Precision()))
+ default:
+ return false
+ }
+}
+
+func validateFilePartitionValues(schema *iceberg.Schema, spec
iceberg.PartitionSpec, values map[int]any) error {
+ if err := validatePartitionValues(spec, values); err != nil {
+ return fmt.Errorf("%w: %v", iceberg.ErrInvalidArgument, err)
+ }
+
+ for _, field := range spec.Fields() {
+ value, supplied := values[field.FieldID]
+ if !supplied {
+ continue
+ }
+
+ switch transform := field.Transform.(type) {
+ case iceberg.BucketTransform:
+ if err := validateFileBucketValue(field, value,
transform.NumBuckets); err != nil {
+ return err
+ }
+ case *iceberg.BucketTransform:
Review Comment:
Partition specs don't contain `*iceberg.BucketTransform` or
`*iceberg.VoidTransform`, so these pointer cases, including the nil check, can
be removed.
##########
table/transaction.go:
##########
@@ -2246,7 +2246,8 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
type AddDataFilesByPathOption func(addFilesOp *addFilesOperation)
type addFilesOperation struct {
- concurrency int
+ concurrency int
+ partitionValues *map[string]map[int]any
Review Comment:
A nil `map[string]map[int]any` already means "not set", so the pointer isn't
needed.
##########
table/file_to_data_file_test.go:
##########
@@ -31,6 +31,9 @@ import (
"github.com/stretchr/testify/require"
)
+// Preserve compatibility for callers that store FileToDataFile as a function
value.
Review Comment:
This only re-checks the existing signature, which `apidiff` already does in
CI, so it can be dropped.
##########
table/arrow_utils.go:
##########
@@ -1853,7 +1853,11 @@ func computeStatsPlan(sc *iceberg.Schema, props
iceberg.Properties) (result map[
return result, nil
}
-func filesToDataFiles(ctx context.Context, fileIO iceio.IO, meta
*MetadataBuilder, filePaths []string, concurrency int) (_ []iceberg.DataFile,
err error) {
+func filesToDataFiles(ctx context.Context, fileIO iceio.IO, meta
*MetadataBuilder, filePaths []string, concurrency int, partitionValuesByPath
...map[string]map[int]any) (_ []iceberg.DataFile, err error) {
Review Comment:
These are unexported, so a plain `map[string]map[int]any` parameter (nil
when not set) would be clearer than a variadic argument; the same applies to
`fileToDataFile` at line 2098.
##########
README.md:
##########
@@ -293,6 +293,40 @@ the table, the following tracks the current write support:
| Row Delta | X |
+### Importing bucket-partitioned Parquet files (v1 and newer)
Review Comment:
This would fit better on the docs site than in the top-level README.
--
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]
