[pqarrow] pqarrow.FileWriter.WriteBuffered returns "unknown error type: interface conversion: interface is nil, not encoding.Encoder[int64]"

[alexandre-normand]

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

github.com/apache/arrow-go/v18: v18.5.1

I'm using arrow-go to write arrow parquet files and using WriteBuffered to periodically flush record batches buffered in memory to disk every 10 seconds. It generally works fine but sometimes I get an error on WriteBuffered like this:

unknown error type: interface conversion: interface is nil, not encoding.Encoder[int64]

or

unknown error type: interface conversion: interface is nil, not encoding.Encoder[github.com/apache/arrow-go/v18/parquet.ByteArray]

I tried dumping the record batch on which the WriteBuffered fails to disk as json and the rows all look complete and correct.

While I don't have an easy reproducible test because I can't find the trigger for the issue, the parquet writer is created like this:

fileWriter, err := pqarrow.NewFileWriter(schema.arrowSchema, file, parquet.NewWriterProperties(parquet.WithAllocator(memory.DefaultAllocator)), pqarrow.DefaultWriterProps())
...

And then we call something like this every 10 seconds:

...
                 builder := array.NewRecordBuilder(memory.DefaultAllocator, schema.arrowSchema)
		defer builder.Release()

		arrowBuilder := newArrowBuilder(schema.arrowSchema, builder)
		for _, op := range toWrite {
			aErr = op.appendArrowRecord(arrowBuilder)
			if aErr != nil {
				aErr = ingestion.NewIrrecoverableError(fmt.Errorf("failed to append arrow record for write operation %s: %w", op.key.String(), aErr), "arrow_transformation_failure")
				return aErr
			}
		}

		// Create a new record batch from the buffer we just filled
		batch := builder.NewRecordBatch()
		defer batch.Release()

		aErr = activeBuffer.fileWriter.WriteBuffered(batch)
		if aErr != nil {
			// Dump a json representation of the batch in blob storage for inspection/troubleshooting
			jsonFileLocation, err := dumpBatchJSON(activeBuffer.fs, activeBuffer.locationProvider, batch)
			if err != nil {
				jsonFileLocation = "unavailable_failure_to_dump_batch"
			}
			aErr = fmt.Errorf("failed to write batch to parquet file, see json dump at '%s' for troubleshooting: %w", jsonFileLocation, aErr)
			return aErr
		}
...

That code obviously doesn't compile but it's a decent representation of how we use the APIs. I initially suspected we were misusing arrow-go/pqarrow but the fact that this isn't deterministic given the same data makes me think that there might be a bug.

I will also add that the failure on WriteBuffered seems to leave the file writer in an inconsistent state as I also see occasional failures on Close() like the one below and they seem to correlate with failures on WriteBuffered that happened prior to that file being closed:

row mismatch for buffered row group: 0, column: 60, count expected: 55000, actual: 54397

Note that I could run with a modified version of arrow-go if there are hypotheses to test. Usually, I can get this error within 30 minutes of startup.

Component(s)

Parquet

[zeroshade]

Looking through the code, one thing I've noticed is that we don't actually validate that the schema of the recordbatch matches the schema of the parquet file being written. In that scenario, if the schemas don't line up or if the record batch has more columns than is expected etc. that could result in the error you're encountering. For example, it attempts to load the encoder for a given int64 or string/binary column in the record batch, but there's no corresponding leaf column in the parquet schema. In that situation the encoder would be nil instead of the expected type.

Can you try dumping the actual schema of the record batch being written at the point you encounter this error and validate it against the schema that was used when you started the file? Make sure they are exactly identical.

As for the error you're encountering on the failure of WriteBuffered, that would be because it failed to write values to a particular column of the row group while writing values to the columns before that. But parquet requires ALL column chunks in a row group to have the same number of rows. So if it failed writing a chunk of rows to a column because of the failed WriteBuffered, the check performed when closing the file would trigger the error you're seeing that there are fewer rows in that column than in the metadata and in the other rows.

This is probably a case where we can improve the error output to better report what is going on to the user.

[zeroshade]

Hmm, looks like i might be wrong. We do have a schema check, but is it possible you're writing to a FileWriter after it has been Closed? It looks like WriteBuffered doesn't actually check fw.closed and depending on the right sequence of events, that could lead to this error message...

[alexandre-normand]

Hmm, looks like i might be wrong. We do have a schema check, but is it possible you're writing to a FileWriter after it has been Closed? It looks like WriteBuffered doesn't actually check fw.closed and depending on the right sequence of events, that could lead to this error message

ah yes, that's interesting. We found a race condition where this could be happening now and we're fixing it. I should have results in a bit.

[alexandre-normand]

It looks like that was it. With the race condition fixed, we haven't seen a single occurrence of the error. @zeroshade : It seems like it would be an easy fix to check if the writer is already closed on Write/WriteBuffered and return an error. Should I go ahead and do that?

[zeroshade]

Yup please do!