nssalian commented on code in PR #2002:
URL: https://github.com/apache/iceberg-go/pull/2002#discussion_r4051817469
##########
table/internal/parquet_files.go:
##########
@@ -711,9 +711,97 @@ func getWriteProperties(writeProps any, arrowSchema
*arrow.Schema) (*parquet.Wri
wp = append(wp, parquet.WithStoreDecimalAsInteger(true))
}
+ // Match Iceberg Java: apply parquet-mr's cost-based dictionary
fallback to every leaf
+ // column so high-cardinality columns fall back to PLAIN rather than
keeping a dictionary.
+ // arrow-go otherwise enables it only for uncompressed columns, so zstd
(our default) would
+ // retain dictionaries on all-distinct columns and roughly double their
size.
+ costFallback, err := dictCostFallbackProps(arrowSchema, wp)
+ if err != nil {
+ return nil, err
+ }
+ wp = append(wp, costFallback...)
+
return parquet.NewWriterProperties(wp...), nil
}
+// dictCostFallbackProps returns a WithDictionaryCostFallbackFor(true)
property per leaf, walking the arrow schema directly (extensions unwrapped) and
falling back to pqarrow.ToParquet for list/map schemas.
+func dictCostFallbackProps(arrowSchema *arrow.Schema, base
[]parquet.WriterProperty) ([]parquet.WriterProperty, error) {
+ if schemaHasListOrMap(arrowSchema) {
+ return dictCostFallbackViaParquet(arrowSchema, base)
+ }
+
+ var props []parquet.WriterProperty
+ var walk func(prefix string, dt arrow.DataType)
+ walk = func(prefix string, dt arrow.DataType) {
+ if ext, ok := dt.(arrow.ExtensionType); ok {
+ dt = ext.StorageType()
+ }
+ if d, ok := dt.(*arrow.DictionaryType); ok {
+ dt = d.ValueType
+ }
+ if st, ok := dt.(*arrow.StructType); ok {
+ for _, f := range st.Fields() {
+ walk(prefix+"."+f.Name, f.Type)
+ }
+
+ return
+ }
+ props = append(props,
parquet.WithDictionaryCostFallbackFor(prefix, true))
+ }
+ for _, f := range arrowSchema.Fields() {
+ walk(f.Name, f.Type)
+ }
+
+ return props, nil
+}
+
+// dictCostFallbackViaParquet is the authoritative path for list/map schemas,
whose parquet leaf naming the direct walk does not reproduce.
+func dictCostFallbackViaParquet(arrowSchema *arrow.Schema, base
[]parquet.WriterProperty) ([]parquet.WriterProperty, error) {
+ parquetSchema, err := pqarrow.ToParquet(arrowSchema,
parquet.NewWriterProperties(base...), pqarrow.DefaultWriterProps())
+ if err != nil {
+ return nil, err
+ }
+
+ props := make([]parquet.WriterProperty, 0, parquetSchema.NumColumns())
+ for i := range parquetSchema.NumColumns() {
+ props = append(props,
parquet.WithDictionaryCostFallbackFor(parquetSchema.Column(i).Path(), true))
+ }
+
+ return props, nil
+}
+
+func schemaHasListOrMap(sc *arrow.Schema) bool {
+ var has func(dt arrow.DataType) bool
+ has = func(dt arrow.DataType) bool {
+ if ext, ok := dt.(arrow.ExtensionType); ok {
+ dt = ext.StorageType()
+ }
+ if d, ok := dt.(*arrow.DictionaryType); ok {
+ dt = d.ValueType
+ }
+ switch t := dt.(type) {
+ case *arrow.ListType, *arrow.LargeListType,
*arrow.FixedSizeListType,
Review Comment:
Done - both `schemaHasListOrMap` and the dict-fallback walk now unwrap
`*arrow.RunEndEncodedType` via `.Encoded()`, after the dictionary unwrap.
--
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]
