Difference between arrow-go and pyarrow file writer

[Dirt-Nasty]

Describe the bug, including details regarding any error messages, version, and platform.

I've decided to use arrow-go to add functionality to my application that takes a GRPC stream and writes it out to parquet. I then use these parquets to load into Snowflake tables as part of our historical ETL processing. I noticed once loaded in that there was misplaced data. There were values on some rows that belonged to others. Primarily on array structs.

After beating my head into the wall debugging this I decided to repack the parquet using python/pyarrow. To my surprise this allowed data to land in the correct place upon ingestion.

I really dont want to have to write this into the processing and move from Go -> Python before landing into blob storage for ingestion.

I did a search here on issues but didn't find anyone reporting a similar issue.

Any insight into some things I might be able to try would be great.

Component(s)

Parquet

[amoeba]

Hi @Dirt-Nasty, a minimal reproducer would be the best next step. Short of that, as much simplified code as you can share could help.

[Dirt-Nasty]

@amoeba

Hard to reproduce without access to a Snowflake environment.

My writer on the go side is pretty large so

This is a small example code block of how I use the go writer:

package main

import (
	"os"

	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/memory"
	"github.com/apache/arrow-go/v18/parquet"
	"github.com/apache/arrow-go/v18/parquet/compress"
	"github.com/apache/arrow-go/v18/parquet/pqarrow"
)

func main() {
	pool := memory.NewGoAllocator()

	opsStruct := arrow.StructOf(
		arrow.Field{Name: "id", Type: arrow.BinaryTypes.String, Nullable: false},
		arrow.Field{Name: "token", Type: arrow.BinaryTypes.String, Nullable: true},
		arrow.Field{Name: "amount", Type: arrow.BinaryTypes.String, Nullable: true},
	)

	schema := arrow.NewSchema([]arrow.Field{
		{Name: "block_num", Type: arrow.PrimitiveTypes.Uint64, Nullable: false},
		{Name: "tx_id", Type: arrow.BinaryTypes.String, Nullable: false},
		// Repro target: array<struct<...>>
		{Name: "ops", Type: arrow.ListOf(opsStruct), Nullable: true},
	}, nil)

	b := array.NewRecordBuilder(pool, schema)
	defer b.Release()

	// scalar columns
	b.Field(0).(*array.Uint64Builder).AppendValues([]uint64{100, 101, 102}, nil)
	b.Field(1).(*array.StringBuilder).AppendValues([]string{"tx-a", "tx-b", "tx-c"}, nil)

	// ops: list<struct<id, token, amount>>
	lb := b.Field(2).(*array.ListBuilder)
	sb := lb.ValueBuilder().(*array.StructBuilder)
	idb := sb.FieldBuilder(0).(*array.StringBuilder)
	tokb := sb.FieldBuilder(1).(*array.StringBuilder)
	amtb := sb.FieldBuilder(2).(*array.StringBuilder)

	// row 0 => 2 elements
	lb.Append(true)
	sb.Append(true); idb.Append("op-1"); tokb.Append("USDC"); amtb.Append("10")
	sb.Append(true); idb.Append("op-2"); tokb.Append("ETH");  amtb.Append("1.5")

	// row 1 => empty list
	lb.Append(true)

	// row 2 => 1 element (nullable struct fields populated partially)
	lb.Append(true)
	sb.Append(true); idb.Append("op-3"); tokb.AppendNull(); amtb.Append("42")

	rec := b.NewRecord()
	defer rec.Release()

	f, err := os.Create("go_nested.parquet")
	if err != nil {
		panic(err)
	}
	defer f.Close()

	props := parquet.NewWriterProperties(
		parquet.WithCompression(compress.Codecs.Zstd),
		parquet.WithDictionaryDefault(true),
		parquet.WithStats(true),
		parquet.WithMaxRowGroupLength(10000),
	)
	arrowProps := pqarrow.NewArrowWriterProperties(pqarrow.WithStoreSchema())

	pw, err := pqarrow.NewFileWriter(schema, f, props, arrowProps)
	if err != nil {
		panic(err)
	}
	if err := pw.Write(rec); err != nil {
		panic(err)
	}
	if err := pw.Close(); err != nil {
		panic(err)
	}
}

On the python side after the go writer has completed is only a few lines.

def rewrite_repack(a_path, out_path, batch_size=65536):
    pf_a = pq.ParquetFile(a_path)
    schema_a = pf_a.schema_arrow
    writer = pq.ParquetWriter(out_path, schema_a)
    try:
        for batch in pf_a.iter_batches(batch_size=batch_size):
            writer.write_batch(batch)
    finally:
        writer.close()

[zeroshade]

@Dirt-Nasty Given the example data in your Go writer version, can you describe in what way the data would end up misplaced? It's interesting that simply reading the batches into python and then writing batches back out fixes the issue. This means that pyarrow is able to correctly interpret the data but does something slightly different when writing it back out. I'll see if I can try to track down what the difference between them is. But it would definitely help if you could describe how Snowflake is "misplacing" the data on different rows etc. as that would help narrow down my search.

[Dirt-Nasty]

@zeroshade in the simplest terms the array of structs ends up on the wrong row after the parquet is copied into snowflake. After alot of debugging running it through pyarrow made the array struct placement show on the correct row. When analyzing the parquet generated from the Go library in something like duckdb the data appears on the correct row, its not like that prior to ingestion. Its just some small difference in storage that when unpacked Snowflake likes the pyarrow version better than the go version.

I can share privately a parquet generated by the go library and then the same one repacked with pyarrow if that helps.

[zeroshade]

That would be significantly helpful, yes!

[zeroshade]

So doing a bit of digging, it looks like the only difference I was able to find between the version written by arrow-go and the version re-packed by pyarrow was a schema name mismatch in the list element.

the stored ARROW:schema metadata uses the element field named "item" which is the Arrow convention, which we always rename to "element" in the Parquet schema (which is the Parquet spec condition). As a result, the Parquet column path has ops.list.element.id but the stored ARROW:schema says the element name is "item". When repacking it using pyarrow, pyarrow reconstructs the Arrow schema from the native Parquet schema and uses "element", making the ARROW:schema consistent (by the way, the parquet reader in arrow-go does the same thing, ignoring the stored element name).

So my current theory is that this mismatch is the issue you're seeing in snowflake if it is relying on the ARROW:schema field names. Can you try testing out #746 and seeing if that solves the issue you're seeing?