Fix entire column default initialize for arrow (#2580)

Arrow uses chunked buffers but default_initialize assumed that data is
contiguous, which was causing segfaults.

Adds:
- Test which repro-ed the segfault
- Fix by using `NullValueReducer::finalize`
- Change `Column::default_initialize_rows` to use `bytes_at` which does
correct boundary checks
- Adds a TODO comment to replace all `ptr_cast` calls to use `bytes_at`.
Not done in this PR to avoid risk of touching very low level code.

#### Reference Issues/PRs
<!--Example: Fixes #1234. See also #3456.-->

#### What does this implement or fix?

#### Any other comments?

#### Checklist

<details>
  <summary>
   Checklist for code changes...
  </summary>
 
- [ ] Have you updated the relevant docstrings, documentation and
copyright notice?
- [ ] Is this contribution tested against [all ArcticDB's
features](../docs/mkdocs/docs/technical/contributing.md)?
- [ ] Do all exceptions introduced raise appropriate [error
messages](https://docs.arcticdb.io/error_messages/)?
 - [ ] Are API changes highlighted in the PR description?
- [ ] Is the PR labelled as enhancement or bug so it appears in
autogenerated release notes?
</details>

<!--
Thanks for contributing a Pull Request to ArcticDB! Please ensure you
have taken a look at:
- ArcticDB's Code of Conduct:
https://github.com/man-group/ArcticDB/blob/master/CODE_OF_CONDUCT.md
- ArcticDB's Contribution Licensing:
https://github.com/man-group/ArcticDB/blob/master/docs/mkdocs/docs/technical/contributing.md#contribution-licensing
-->

Author: IvoDD

cpp/arcticdb/column_store/chunked_buffer.hpp

@@ -386,6 +386,9 @@ class ChunkedBufferImpl {
 
     template<typename T>
     T *ptr_cast(size_t pos_bytes, size_t required_bytes) {
+        // TODO: This check doesn't verify we're overreaching outside of block boundaries.
+        // We should instead use `bytes_at` which does the correct check like so:
+        // return reinterpret_cast<T *>(bytes_at(pos_bytes, required_bytes))
         check_bytes(pos_bytes, required_bytes);
         return reinterpret_cast<T *>(&operator[](pos_bytes));
     }

cpp/arcticdb/column_store/column.cpp

@@ -386,7 +386,8 @@ void Column::default_initialize_rows(size_t start_pos, size_t num_rows, bool ens
             if (ensure_alloc) {
                 data_.ensure<uint8_t>(bytes);
             }
-            auto type_ptr = data_.ptr_cast<RawType>(start_pos, bytes);
+            // This doesn't work if we default_initialize bytes which span across multiple blocks.
+            auto type_ptr = reinterpret_cast<RawType *>(data_.bytes_at(start_pos * sizeof(RawType), bytes));
             util::initialize<T>(reinterpret_cast<uint8_t*>(type_ptr), bytes, default_value);
             if (ensure_alloc) {
                 data_.commit();

cpp/arcticdb/pipeline/read_frame.cpp

@@ -925,7 +925,8 @@ struct ReduceColumnTask : async::BaseTask {
                     auto& prev_buffer = column.buffer();
                     swap(prev_buffer, new_buffer);
                 } else {
-                    column.default_initialize_rows(0, frame_.row_count(), false, default_value);
+                    NullValueReducer null_reducer{column, context_, frame_, shared_data_, handler_data_, read_options_.output_format(), default_value};
+                    null_reducer.finalize();
                 }
             }
         } else if (column_data != slice_map_->columns_.end()) {

cpp/arcticdb/util/cursored_buffer.hpp

@@ -141,13 +141,11 @@ struct CursoredBuffer {
     }
 
     uint8_t* bytes_at(size_t bytes, size_t required) {
-        util::check(bytes + required <= buffer_.bytes(), "Bytes overflow, can't write {} bytes at position {} in buffer of size {}", required, bytes, buffer_.bytes());
-        return &buffer_[bytes];
+        return buffer_.bytes_at(bytes, required);
     }
 
     const uint8_t* bytes_at(size_t bytes, size_t required) const {
-        util::check(bytes + required <= buffer_.bytes(), "Bytes overflow, can't write {} bytes at position {} in buffer of size {}", required, bytes, buffer_.bytes());
-        return &buffer_[bytes];
+        return buffer_.bytes_at(bytes, required);
     }
 
     void clear() {

python/tests/unit/arcticdb/version_store/test_arrow.py

@@ -566,3 +566,26 @@ def test_arrow_sparse_floats_hypothesis(lmdb_version_store_arrow, df, rows_per_s
         expected = pa.concat_tables([pa.Table.from_pandas(row_slice) for row_slice in row_slices])
         received = lib.read(sym).data
     assert expected.equals(received)
+
+
+@pytest.mark.parametrize(
+    "type_to_drop", [pa.int64(), pa.float64(), pa.large_string()]
+)
+def test_arrow_dynamic_schema_filtered_column(lmdb_version_store_dynamic_schema_v1, type_to_drop):
+    lib = lmdb_version_store_dynamic_schema_v1
+    lib.set_output_format(OutputFormat.EXPERIMENTAL_ARROW)
+    sym = "sym"
+    column_to_drop = pa.array(["a", "b"], type_to_drop) if type_to_drop == pa.large_string() else pa.array([1, 2], type_to_drop)
+    table_1 = pa.table({"col": pa.array([0, 1])})
+    table_2 = pa.table({"col": pa.array([5, 6]), "col_to_drop": column_to_drop})
+    table_3 = pa.table({"col": pa.array([2, 3])})
+    # TODO: Remove to_pandas() when we support writing Arrow structures directly
+    lib.write(sym, table_1.to_pandas())
+    lib.append(sym, table_2.to_pandas())
+    lib.append(sym, table_3.to_pandas())
+    expected = pa.concat_tables([table_1, table_2, table_3], promote_options="permissive")
+    expected = expected.filter(pa.compute.field("col") < 5)
+    q = QueryBuilder()
+    q = q[q["col"] < 5]
+    received = stringify_dictionary_encoded_columns(lib.read(sym, query_builder=q).data)
+    assert expected.equals(received)