diff --git a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py index 7f06fcab4299..99ecc87bdfa4 100644 --- a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py +++ b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py @@ -16,12 +16,16 @@ # under the License. import os +from contextlib import contextmanager import unittest from unittest import mock import pyarrow as pa +import pyarrow.parquet as pq +from pypaimon.read.table_read import TableRead from pypaimon.table.special_fields import SpecialFields +from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter from pypaimon.tests.data_evolution_test_helpers import ( BatchModeMixin, DataEvolutionTestBase, @@ -61,6 +65,184 @@ def _apply_upsert(self, table_update, data, upsert_keys, cid): def _apply_upsert_rows(self, table_update, rows, upsert_keys, cid): raise NotImplementedError + def test_upsert_row_groups_do_not_follow_read_batches(self): + schema = pa.schema([('id', pa.int32()), ('score', pa.int32())]) + for read_size in (73, 1024): + with self.subTest(read_size=read_size): + table = self._create_table(pa_schema=schema, options={ + **self.table_options, 'read.batch-size': str(read_size)}) + original = pa.Table.from_pydict( + {'id': list(range(5000)), 'score': list(range(5000))}, schema=schema) + self._write_arrow(table, original) + updates = pa.Table.from_pydict({'id': [13], 'score': [-1]}, schema=schema) + messages = self._upsert(table, updates, ['id'], ['score']) + files = [f for message in messages for f in message.new_files] + self.assertEqual(len(files), 1) + metadata = pq.read_metadata(files[0].file_path) + self.assertEqual(metadata.num_row_groups, 1) + self.assertEqual(metadata.row_group(0).num_rows, 5000) + scores = self._read_all(table).to_pydict()['score'] + expected = list(range(5000)) + expected[13] = -1 + self.assertEqual(scores, expected) + + def test_row_group_byte_budget_and_oversized_row(self): + schema = pa.schema([('id', pa.int32())]) + table = self._create_table(pa_schema=schema, options={ + **self.table_options, 'file.block-size': '400 b'}) + writer = AppendOnlyDataWriter(table, (), 0, 0, table.options) + data = pa.Table.from_pydict({'id': list(range(250))}, schema=schema) + try: + for batch_size in (1, 73, 250): + groups = list(writer._row_groups(data.to_batches(max_chunksize=batch_size))) + self.assertEqual([group.num_rows for group in groups], [100, 100, 50]) + self.assertTrue(pa.concat_tables(groups).equals(data)) + self.assertTrue(all(group.nbytes <= 400 for group in groups)) + large = pa.Table.from_pydict({'text': ['a', 'x' * 500, 'b']}) + groups = list(writer._row_groups(large.to_batches())) + self.assertEqual([group.num_rows for group in groups], [1, 1, 1]) + self.assertTrue(pa.concat_tables(groups).equals(large)) + # Empty batches, nulls and sliced variable-width columns must not + # lose rows or make the output depend on input batch boundaries. + mixed = pa.Table.from_pydict({ + 'text': [ + 'skip', None, '', 'a' * 300, + 'b' * 300, None, 'last', 'skip', + ], + }).slice(1, 6) + layouts = [] + for batch_size in (1, 2, 6): + batches = mixed.to_batches(max_chunksize=batch_size) + batches.insert(0, batches[0].slice(0, 0)) + groups = list(writer._row_groups(batches)) + self.assertTrue(pa.concat_tables(groups).equals(mixed)) + self.assertTrue(all(group.nbytes <= 400 for group in groups)) + layouts.append([group.num_rows for group in groups]) + self.assertTrue(all(layout == layouts[0] for layout in layouts)) + self.assertEqual(list(writer._row_groups([])), []) + with mock.patch.object(AppendOnlyDataWriter, '_ROW_GROUP_MAX_ROWS', 17): + groups = list(writer._row_groups(data.to_batches(max_chunksize=73))) + self.assertEqual([group.num_rows for group in groups], [17] * 14 + [12]) + self.assertTrue(pa.concat_tables(groups).equals(data)) + finally: + writer.close() + + @mock.patch.object(AppendOnlyDataWriter, '_ROW_GROUP_MAX_ROWS', 2) + def test_partial_upsert_streams_original_file_group(self): + schema = pa.schema([ + ('id', pa.int32()), + ('payload', pa.list_(pa.struct([('text', pa.string())]))), + ('score', pa.int32()), + ]) + table = self._create_table(pa_schema=schema, options={ + **self.table_options, 'metadata.stats-mode': 'full'}) + expected = [{'id': i, 'payload': [{'text': str(i)}], 'score': i} + for i in range(12)] + + def as_table(rows): + return pa.Table.from_pydict( + {name: [row[name] for row in rows] for name in schema.names}, + schema=schema) + + self._write_arrow(table, as_table(expected)) + read_batches = TableRead._to_managed_arrow_batch_reader + write = pq.ParquetWriter.write_table + progress = {'read': 0, 'written': 0} + closed = [] + + @contextmanager + def bounded_reader(reader, splits, **kwargs): + source = read_batches(reader, splits, **kwargs) + + def batches(): + try: + for batch in source: + for start in range(0, batch.num_rows, 2): + self.assertEqual(progress['read'], progress['written']) + piece = batch.slice(start, 2) + progress['read'] += piece.num_rows + yield piece + finally: + source.close() + closed.append(True) + iterator = batches() + try: + yield iterator + finally: + iterator.close() + + def record_write(writer, batch, **kwargs): + result = write(writer, batch, **kwargs) + progress['written'] += batch.num_rows + return result + + for replacements in ([9, 1, 5], [0, 11]): + updates = [{'id': i, 'payload': None if i == 5 else + [{'text': 'updated-' + str(i)}], + 'score': None if i == 5 else -i} for i in replacements] + progress.update(read=0, written=0) + with mock.patch.object(TableRead, 'to_arrow', side_effect=AssertionError( + 'upsert must not materialize the original file group')): + with mock.patch.object(TableRead, '_to_managed_arrow_batch_reader', bounded_reader): + with mock.patch.object(pq.ParquetWriter, 'write_table', record_write): + messages = self._upsert( + table, as_table(updates), + ['id'], ['payload', 'score']) + self.assertEqual(progress, {'read': 12, 'written': 12}) + for row in updates: + expected[row['id']] = row + self.assertEqual(self._read_all(table).to_pydict(), as_table(expected).to_pydict()) + files = [f for msg in messages for f in msg.new_files] + self.assertEqual(len(files), 1) + self.assertEqual((files[0].first_row_id, files[0].row_count), (0, 12)) + scores = [r['score'] for r in expected if r['score'] is not None] + self.assertEqual(files[0].value_stats.min_values.values[1], min(scores)) + self.assertEqual(files[0].value_stats.max_values.values[1], max(scores)) + self.assertEqual(files[0].value_stats.null_counts, [1, 1]) + + # A failed streamed write must leave the committed table intact and + # remove the partially written overlay. + self.assertEqual(len(closed), 2) + progress.update(read=0, written=0) + + def fail_second_write(writer, batch, **kwargs): + if progress['written']: + raise OSError('injected write failure') + return record_write(writer, batch, **kwargs) + + with mock.patch.object(table.file_io, 'delete_quietly', + wraps=table.file_io.delete_quietly) as delete: + with mock.patch.object(pq.ParquetWriter, 'write_table', + fail_second_write): + with mock.patch.object(TableRead, '_to_managed_arrow_batch_reader', bounded_reader): + with self.assertRaisesRegex(OSError, 'injected write failure'): + self._upsert(table, as_table(updates), + ['id'], ['payload', 'score']) + self.assertEqual(len(closed), 3) + self.assertTrue(delete.called) + for call in delete.call_args_list: + self.assertFalse(table.file_io.exists(call[0][0])) + self.assertEqual(self._read_all(table).to_pydict(), as_table(expected).to_pydict()) + + # Closing the format writer is part of the file transaction too. + close = pq.ParquetWriter.close + + def fail_close(writer): + was_open = writer.is_open + close(writer) + if was_open: + raise OSError('injected close failure') + + with mock.patch.object(table.file_io, 'delete_quietly', + wraps=table.file_io.delete_quietly) as delete: + with mock.patch.object(pq.ParquetWriter, 'close', fail_close): + with self.assertRaisesRegex(OSError, 'injected close failure'): + self._upsert(table, as_table(updates), ['id'], ['payload', 'score']) + self.assertTrue(delete.called) + for call in delete.call_args_list: + self.assertFalse(table.file_io.exists(call[0][0])) + self.assertEqual(self._read_all(table).to_pydict(), as_table(expected).to_pydict()) + # ------------------------------------------------------------------ # Helpers built on the primitives # ------------------------------------------------------------------ @@ -953,6 +1135,61 @@ def test_update_cols_partial_update(self): self.assertEqual((1, 'Alice', 99, 'NYC'), rows[0]) self.assertEqual((2, 'Bob', 88, 'LA'), rows[1]) + def test_duplicate_update_cols_are_deduplicated(self): + table = self._create_table() + self._write_arrow(table, pa.Table.from_pydict({ + 'id': [1], + 'name': ['Alice'], + 'age': [25], + 'city': ['NYC'], + }, schema=self.pa_schema)) + + messages = self._upsert( + table, + pa.Table.from_pydict({ + 'id': [1], + 'name': ['ignored'], + 'age': [99], + 'city': ['ignored'], + }, schema=self.pa_schema), + upsert_keys=['id'], + # Matching the schema width must not mean "update all columns". + update_cols=['age'] * len(table.field_names), + ) + + self.assertEqual( + self._read_all(table).to_pydict(), + {'id': [1], 'name': ['Alice'], 'age': [99], 'city': ['NYC']}, + ) + files = [file for message in messages for file in message.new_files] + self.assertEqual([file.write_cols for file in files], [['age']]) + + def test_not_null_update_across_read_batches(self): + schema = pa.schema([ + pa.field('id', pa.int32(), nullable=False), + pa.field('score', pa.int32(), nullable=False), + ]) + table = self._create_table(pa_schema=schema, options={ + **self.table_options, 'read.batch-size': '2'}) + original = pa.Table.from_pydict({ + 'id': list(range(4)), + 'score': list(range(4)), + }, schema=schema) + self._write_arrow(table, original) + + updates = pa.Table.from_pydict({'id': [2], 'score': [99]}, schema=schema) + messages = self._upsert(table, updates, ['id'], ['score']) + + expected = original.set_column( + 1, + schema.field('score'), + pa.array([0, 1, 99, 3], type=pa.int32()), + ) + self.assertTrue(self._read_all(table).equals(expected)) + files = [file for message in messages for file in message.new_files] + self.assertEqual(len(files), 1) + self.assertFalse(pq.read_schema(files[0].file_path).field('score').nullable) + # ================================================================== # Duplicate-key dedup tests — parametrised # ================================================================== diff --git a/paimon-python/pypaimon/write/table_update.py b/paimon-python/pypaimon/write/table_update.py index c358644460ac..922b1bc8c33b 100644 --- a/paimon-python/pypaimon/write/table_update.py +++ b/paimon-python/pypaimon/write/table_update.py @@ -126,6 +126,7 @@ def __init__(self, table, commit_user): self.projection = None def with_update_type(self, update_cols: List[str]): + update_cols = list(dict.fromkeys(update_cols)) for col in update_cols: if col not in self.table.field_names: raise ValueError(f"Column {col} is not in table schema.") diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py b/paimon-python/pypaimon/write/table_update_by_row_id.py index 90c08d20d2aa..c0d582d48c80 100644 --- a/paimon-python/pypaimon/write/table_update_by_row_id.py +++ b/paimon-python/pypaimon/write/table_update_by_row_id.py @@ -23,6 +23,7 @@ import pyarrow as pa import pyarrow.compute as pc +from pypaimon.common.options.core_options import ChangelogProducer from pypaimon.manifest.schema.data_file_meta import DataFileMeta from pypaimon.manifest.schema.manifest_entry import ManifestEntry from pypaimon.read.scanner.data_evolution_split_generator import ( @@ -49,6 +50,7 @@ value_for_arrow, ) from pypaimon.write.writer.blob_writer import BlobWriter +from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter @dataclass(frozen=True) @@ -211,6 +213,7 @@ def update_columns(self, data: pa.Table, column_names: List[str]) -> List[Commit if not column_names: raise ValueError("column_names cannot be empty") + column_names = list(dict.fromkeys(column_names)) if SpecialFields.ROW_ID.name not in data.column_names: raise ValueError(f"Input data must contain {SpecialFields.ROW_ID.name} column") @@ -257,6 +260,7 @@ def update_rows_columns( ) -> List[CommitMessage]: if not column_names: raise ValueError("column_names cannot be empty") + column_names = list(dict.fromkeys(column_names)) if len(rows) != len(row_ids_by_row): raise ValueError( "rows and row_ids_by_row must have the same length: " @@ -390,7 +394,7 @@ def _write_by_first_row_id( group_blob_object_columns, ) - def _read_original_file_data(self, first_row_id: int, column_names: List[str]) -> Optional[pa.Table]: + def _read_original_file_data(self, first_row_id: int, column_names: List[str]) -> pa.Table: """Read original file data for the given first_row_id. Only reads columns that exist in the original file and need to be updated. @@ -402,16 +406,17 @@ def _read_original_file_data(self, first_row_id: int, column_names: List[str]) - column_names: The column names to update Returns: - PyArrow Table containing the original data for columns that exist in the file, - or None if no columns need to be read from the original file. + PyArrow Table containing the original values for the requested columns. """ + table_read, origin_split = self._original_file_read(first_row_id, column_names) + original = table_read.to_arrow([origin_split]) + return original.select(column_names) + + def _original_file_read(self, first_row_id, column_names): wanted = set(column_names) read_fields: List[DataField] = [ table_field for table_field in self.table.fields if table_field.name in wanted ] - if not read_fields: - return None - entry = self._first_row_id_index.get(first_row_id) if entry is None: raise ValueError(f"No file found for first_row_id {first_row_id}") @@ -432,8 +437,52 @@ def _read_original_file_data(self, first_row_id: int, column_names: List[str]) - predicate=None, read_type=read_fields + [SpecialFields.ROW_ID], ) - original = table_read.to_arrow([origin_split]) - return original.select([field.name for field in read_fields]) + return table_read, origin_split + + def _merged_batches(self, first_row_id, data, column_names): + """Merge ordinary columns a batch at a time in physical row order.""" + table_read, split = self._original_file_read(first_row_id, column_names) + updates = sorted(enumerate(data[SpecialFields.ROW_ID.name].to_pylist()), + key=lambda item: item[1]) + update_index = 0 + offset = first_row_id + with table_read._to_managed_arrow_batch_reader([split]) as reader: + for batch in reader: + end = offset + batch.num_rows + selected = [] + while update_index < len(updates) and updates[update_index][1] < end: + selected.append(updates[update_index][0]) + update_index += 1 + original = pa.Table.from_batches([batch]).select(column_names) + if selected: + merged, _ = self._merge_update_with_original( + original, data.take(selected), column_names, offset) + else: + merged = original + yield from merged.to_batches() + offset = end + del batch, original, merged + if update_index != len(updates): + raise ValueError('Update row IDs extend past the original file group') + + def _write_group_streaming(self, partition, first_row_id, data, column_names): + writer = AppendOnlyDataWriter( + self.table, tuple(partition.values), 0, 0, + self.table.options, write_cols=column_names) + batches = self._merged_batches(first_row_id, data, column_names) + try: + files = writer._write_batches(batches) + self._assign_update_file_metadata(files, first_row_id, column_names, {}) + if files: + self.commit_messages.append(CommitMessage( + partition=tuple(partition.values), bucket=0, new_files=files, + check_from_snapshot=self.snapshot_id)) + except Exception: + writer.abort() + raise + finally: + batches.close() + writer.close() def _merge_update_with_original( self, @@ -533,7 +582,13 @@ def _merge_update_with_original( merged_columns[col_name] = self._merge_chunked_column( original_col, update_col, sorted_updates) - merged_table = pa.table(merged_columns) if merged_columns else None + merged_table = None + if merged_columns: + merged_schema = pa.schema([ + original_data.schema.field(name) + for name in merged_columns + ]) + merged_table = pa.table(merged_columns, schema=merged_schema) return merged_table, blob_columns @@ -770,6 +825,18 @@ def _write_group( Reads the original file data, merges in the update values, and writes a single output file (rolling disabled) for the group. """ + options = self.table.options + # Specialized writers still own their sidecars and physical encoding. + if (not self.table.is_primary_key_table + and options.file_format('parquet') == 'parquet' + and not (options.variant_shredding_enabled() + and options.variant_shredding_schema()) + and not options.data_evolution_row_sidecar_enabled(False) + and not options.with_vector_format() + and options.changelog_producer() == ChangelogProducer.NONE + and not any(is_blob_file_field(f) for f in self.table.fields)): + self._write_group_streaming(partition, first_row_id, data, column_names) + return original_data = self._read_original_file_data(first_row_id, column_names) _, target_files = self._first_row_id_index[first_row_id] blob_columns_with_baseline = { diff --git a/paimon-python/pypaimon/write/writer/append_only_data_writer.py b/paimon-python/pypaimon/write/writer/append_only_data_writer.py index 1a4024104fbf..267db327d604 100644 --- a/paimon-python/pypaimon/write/writer/append_only_data_writer.py +++ b/paimon-python/pypaimon/write/writer/append_only_data_writer.py @@ -16,15 +16,142 @@ # under the License. import pyarrow as pa +import pyarrow.parquet as pq +import uuid +from pypaimon.manifest.schema.simple_stats import SimpleStats +from pypaimon.schema.data_types import PyarrowFieldParser +from pypaimon.table.row.generic_row import GenericRow from pypaimon.write.writer.data_writer import DataWriter +from pypaimon.write.writer.write_buffer import WriteBuffer class AppendOnlyDataWriter(DataWriter): """Data writer for append-only tables.""" + _ROW_GROUP_MAX_ROWS = 1024 * 1024 + def _process_data(self, data: pa.RecordBatch) -> pa.Table: return pa.Table.from_batches([data]) def _merge_data(self, existing_data: pa.Table, new_data: pa.Table) -> pa.Table: return pa.concat_tables([existing_data, new_data]) + + @staticmethod + def _row_group_slice(batch, offset, count): + piece = batch.slice(offset, count) + # Arrow 6 nbytes counts full backing buffers even for slices. Compact + # the slice there so both accounting and retained buffers stay bounded. + if int(pa.__version__.split('.')[0]) < 7: + piece = pa.RecordBatch.from_arrays( + [pa.concat_arrays([column]) for column in piece.columns], schema=piece.schema) + return piece + + def _row_groups(self, batches): + """Bound row-group buffering independently of reader batch boundaries. + + Arrow bytes are an estimate, not the encoded Parquet block size. + One oversized row and the current input batch can exceed the target. + """ + configured = self.options.file_block_size() + target_bytes = configured.get_bytes() if configured is not None else 128 * 1024 * 1024 + if target_bytes <= 0: + raise ValueError('file.block-size must be positive') + buffer = WriteBuffer(self._merge_data) + try: + for batch in batches: + offset = 0 + while offset < batch.num_rows: + count = min(batch.num_rows - offset, + self._ROW_GROUP_MAX_ROWS - buffer.num_rows) + piece = self._row_group_slice(batch, offset, count) + available = target_bytes - buffer.nbytes + if piece.nbytes > available: + low, high = 0, count + while low < high: + middle = (low + high + 1) // 2 + if self._row_group_slice(piece, 0, middle).nbytes <= available: + low = middle + else: + high = middle - 1 + count = low + if count == 0 and buffer.num_rows: + yield buffer.take() + continue + count = max(1, count) + piece = self._row_group_slice(piece, 0, count) + buffer.append(pa.Table.from_batches([piece])) + offset += count + del piece + if buffer.nbytes >= target_bytes or buffer.num_rows >= self._ROW_GROUP_MAX_ROWS: + yield buffer.take() + del batch + if buffer.num_rows: + yield buffer.take() + finally: + buffer.reset() + + def _write_batches(self, batches): + """Write one overlay file with bounded row-group buffering. + + Used for ordinary Parquet column updates, without sidecars or + shredding. Keep file boundaries while bounding payload memory. + """ + file_name = '{}{}-0.parquet'.format( + self.options.data_file_prefix(), uuid.uuid4()) + file_path = self._generate_file_path(file_name) + row_count = 0 + stats = {} + fields = [] + + kwargs = {'compression': self.compression} + if self.compression.lower() == 'zstd': + kwargs['compression_level'] = self.zstd_level + groups = self._row_groups(batches) + try: + # Like SingleFileWriter, own the format writer and its output stream. + # FileIO supplies storage access, not the format writer lifecycle. + with self.file_io.new_output_stream(file_path) as stream: + writer = None + try: + for batch in groups: + if not batch.num_rows: + continue + if writer is None: + writer = pq.ParquetWriter(stream, batch.schema, **kwargs) + if self.options.metadata_stats_enabled(): + fields = PyarrowFieldParser.to_paimon_schema(batch.schema) + writer.write_table(batch, row_group_size=batch.num_rows) + row_count += batch.num_rows + for field in fields: + current = self._get_column_stats(batch, field.name) + previous = stats.get(field.name) + if previous is not None: + current['null_counts'] += previous['null_counts'] + for key, choose in (('min_values', min), ('max_values', max)): + values = [v for v in (previous[key], current[key]) if v is not None] + current[key] = choose(values) if values else None + stats[field.name] = current + del batch + finally: + if writer is not None: + writer.close() + if not row_count: + self.file_io.delete_quietly(file_path) + return [] + meta = self._create_data_file_meta( + file_name=file_name, + file_path=file_path, + row_count=row_count, + min_key=GenericRow([], []), max_key=GenericRow([], []), + key_stats=SimpleStats.empty_stats(), + value_stats=self._collect_value_stats(None, fields, stats), + min_sequence_number=0, max_sequence_number=0, + ) + self.committed_files.append(meta) + return [meta] + except Exception: + self.file_io.delete_quietly(file_path) + raise + finally: + groups.close() diff --git a/paimon-python/pypaimon/write/writer/data_writer.py b/paimon-python/pypaimon/write/writer/data_writer.py index 24a67186c3d8..7fe9af4355ee 100644 --- a/paimon-python/pypaimon/write/writer/data_writer.py +++ b/paimon-python/pypaimon/write/writer/data_writer.py @@ -332,9 +332,9 @@ def _write_data_to_file(self, data: pa.Table): min_seq = self.sequence_generator.start max_seq = self.sequence_generator.current creation_time = Timestamp.now() - data_meta = DataFileMeta.create( + data_meta = self._create_data_file_meta( file_name=file_name, - file_size=self.file_io.get_file_size(file_path), + file_path=file_path, row_count=data.num_rows, min_key=GenericRow(min_key, self.trimmed_primary_keys_fields), max_key=GenericRow(max_key, self.trimmed_primary_keys_fields), @@ -342,17 +342,8 @@ def _write_data_to_file(self, data: pa.Table): value_stats=value_stats, min_sequence_number=min_seq, max_sequence_number=max_seq, - schema_id=self.table.table_schema.id, - level=0, extra_files=extra_files, creation_time=creation_time, - delete_row_count=0, - file_source=0, - value_stats_cols=None if value_stats_enabled else [], - external_path=external_path_str, - first_row_id=None, - write_cols=self.write_cols, - file_path=file_path, ) if self.changelog_producer == ChangelogProducer.INPUT: @@ -372,6 +363,28 @@ def _write_data_to_file(self, data: pa.Table): if changelog_meta is not None: self.committed_changelog_files.append(changelog_meta) + def _create_data_file_meta(self, file_name, file_path, row_count, + min_key, max_key, key_stats, value_stats, + min_sequence_number, max_sequence_number, + extra_files=None, creation_time=None): + """Common metadata finalization for buffered and incremental files.""" + return DataFileMeta.create( + file_name=file_name, + file_size=self.file_io.get_file_size(file_path), + row_count=row_count, + min_key=min_key, max_key=max_key, + key_stats=key_stats, value_stats=value_stats, + min_sequence_number=min_sequence_number, + max_sequence_number=max_sequence_number, + schema_id=self.table.table_schema.id, level=0, + extra_files=extra_files if extra_files is not None else [], + creation_time=creation_time if creation_time is not None else Timestamp.now(), + delete_row_count=0, file_source=0, + value_stats_cols=None if self.options.metadata_stats_enabled() else [], + external_path=file_path if self.external_path_provider is not None else None, + first_row_id=None, write_cols=self.write_cols, file_path=file_path, + ) + def _apply_variant_shredding(self, data: pa.Table) -> pa.Table: """Transform VARIANT columns into shredded Parquet format.