Fix segfault writing to no longer valid column writer

cpp/RowGroupWriter.cpp

@@ -47,4 +47,9 @@ extern "C"
 	{
 		TRYCATCH(*total_compressed_bytes = row_group_writer->total_compressed_bytes();)
 	}
+
+	PARQUETSHARP_EXPORT ExceptionInfo* RowGroupWriter_Buffered(const RowGroupWriter* row_group_writer, bool* buffered)
+	{
+		TRYCATCH(*buffered = row_group_writer->buffered();)
+	}
 }

csharp.test/TestLogicalColumnWriter.cs

@@ -27,5 +27,29 @@ public static void TestInvalidCastErrorMessage()
 
             writer.Close();
         }
+
+        [Test]
+        public static void TestWriteAfterNextColumn()
+        {
+            var schemaColumns = new Column[]
+            {
+                new Column<int>("A"),
+                new Column<float>("B"),
+            };
+
+            using var buffer = new ResizableBuffer();
+            using var outStream = new BufferOutputStream(buffer);
+            using var writer = new ParquetFileWriter(outStream, schemaColumns);
+            using var rowGroupWriter = writer.AppendRowGroup();
+
+            using var colWriterA = rowGroupWriter.NextColumn().LogicalWriter<int>();
+            using var colWriterB = rowGroupWriter.NextColumn().LogicalWriter<float>();
+
+            var exception = Assert.Throws<Exception>(() => { colWriterA.WriteBatch(new[] {0, 1, 2, 3, 4}); });
+            Assert.That(exception!.Message, Is.EqualTo(
+                "Writer for column 0 is no longer valid, the current column for the row group writer is 1"));
+
+            writer.Close();
+        }
     }
 }

csharp/ColumnWriter.cs

@@ -37,7 +37,7 @@ internal static ColumnWriter Create(IntPtr handle, RowGroupWriter rowGroupWriter
 
         internal ColumnWriter(IntPtr handle, RowGroupWriter rowGroupWriter, int columnIndex)
         {
-            Handle = handle;
+            _handle = handle;
             RowGroupWriter = rowGroupWriter;
             ColumnIndex = columnIndex;
         }
@@ -159,7 +159,25 @@ protected static extern unsafe IntPtr TypedColumnWriter_WriteBatchSpaced_ByteArr
         protected static extern unsafe IntPtr TypedColumnWriter_WriteBatchSpaced_FixedLenByteArray(
             IntPtr columnWriter, long numValues, short* defLevels, short* repLevels, byte* validBits, long validBitsOffset, FixedLenByteArray* values);
 
-        protected readonly IntPtr Handle;
+        protected IntPtr Handle
+        {
+            get
+            {
+                if (!RowGroupWriter.Buffered)
+                {
+                    var currentColumn = RowGroupWriter.CurrentColumn;
+                    if (ColumnIndex != currentColumn)
+                    {
+                        throw new Exception($"Writer for column {ColumnIndex} is no longer valid, " +
+                                            $"the current column for the row group writer is {currentColumn}");
+                    }
+                }
+
+                return _handle;
+            }
+        }
+
+        private readonly IntPtr _handle;
         internal readonly RowGroupWriter RowGroupWriter;
     }
 

csharp/RowGroupWriter.cs

@@ -27,6 +27,7 @@ public void Close()
         public long NumRows => ExceptionInfo.Return<long>(_handle, RowGroupWriter_Num_Rows);
         public long TotalBytesWritten => ExceptionInfo.Return<long>(_handle, RowGroupWriter_Total_Bytes_Written);
         public long TotalCompressedBytes => ExceptionInfo.Return<long>(_handle, RowGroupWriter_Total_Compressed_Bytes);
+        public bool Buffered => ExceptionInfo.Return<bool>(_handle, RowGroupWriter_Buffered);
 
         public ColumnWriter Column(int i)
         {
@@ -62,6 +63,9 @@ public ColumnWriter NextColumn()
         [DllImport(ParquetDll.Name)]
         private static extern IntPtr RowGroupWriter_Total_Compressed_Bytes(IntPtr rowGroupWriter, out long totalCompressedBytes);
 
+        [DllImport(ParquetDll.Name)]
+        private static extern IntPtr RowGroupWriter_Buffered(IntPtr rowGroupWriter, out bool buffered);
+
         private readonly IntPtr _handle;
         internal readonly ParquetFileWriter ParquetFileWriter;
     }