This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b96b7a6a28 [python] Complete object RowKind filtering and reuse Arrow
event columns (#10462)
b96b7a6a28 is described below
commit b96b7a6a284bfd48cc052b258be5380e9cfe3430
Author: zhigang <[email protected]>
AuthorDate: Fri Oct 9 21:54:48 2026 +0800
[python] Complete object RowKind filtering and reuse Arrow event columns
(#10462)
---
paimon-python/README.md | 2 +
.../pypaimon/tests/write/object_row_kind_test.py | 358 +++++++++++++++++++++
.../tests/write/postpone_event_write_test.py | 14 +-
.../pypaimon/write/postpone_batch_table_write.py | 4 +
paimon-python/pypaimon/write/row_kind.py | 65 ++--
paimon-python/pypaimon/write/writer/data_writer.py | 8 +-
.../pypaimon/write/writer/key_value_data_writer.py | 10 +-
7 files changed, 433 insertions(+), 28 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index f4dd3c7fe8..cc28e3b66d 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -168,6 +168,8 @@ path select the Python writer before native data is
written. If the runtime or t
unavailable, write uses Python. Once Rust starts writing a batch, errors
propagate without retrying that batch through Python.
+For primary-key Row writes, `rowkind.field` takes precedence over the object's
`RowKind`; otherwise the object event is preserved. Plain Arrow input defaults
to INSERT. Both Python and native writers apply `ignore-delete` and
`ignore-update-before` before routing object rows.
+
Batch and stream `merge_into` also use Rust core for eligible data-evolution
Parquet tables when `write.native.enabled=true`. Existing `WhenMatched` and
`WhenNotMatched` clauses accept Arrow/pandas input, Paimon table sources, or
diff --git a/paimon-python/pypaimon/tests/write/object_row_kind_test.py
b/paimon-python/pypaimon/tests/write/object_row_kind_test.py
new file mode 100644
index 0000000000..8158f5d5a9
--- /dev/null
+++ b/paimon-python/pypaimon/tests/write/object_row_kind_test.py
@@ -0,0 +1,358 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Object events survive filtering, routing and Row-to-Arrow conversion."""
+
+from typing import Any, List
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.table.file_store_table import FileStoreTable
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.projected_row import ProjectedRow
+from pypaimon.table.row.row_kind import RowKind
+from pypaimon.write.native_write import NativeTableWrite
+
+
[email protected](
+ params=[
+ pytest.param(False, marks=pytest.mark.python_write, id='python'),
+ pytest.param(True, marks=pytest.mark.native_plan, id='native'),
+ ]
+)
+def native(request):
+ return request.param
+
+
+def make_table(tmp_path, native, bucket='1', options=None, partition=False):
+ fields: List[pa.Field] = [
+ pa.field('id', pa.int32(), nullable=False),
+ pa.field('v', pa.int32()),
+ ]
+ if options and options.get('rowkind.field') is not None:
+ fields.append(pa.field('op', pa.string()))
+ if partition:
+ fields.append(pa.field('pt', pa.string(), nullable=False))
+ schema = pa.schema(fields)
+ settings = {
+ 'bucket': bucket,
+ 'file.format': 'parquet',
+ 'write.native.enabled': str(native).lower(),
+ 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false',
+ }
+ settings.update(options or {})
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table(
+ 'default.events',
+ Schema.from_pyarrow_schema(
+ schema,
+ primary_keys=['id', 'pt'] if partition else ['id'],
+ partition_keys=['pt'] if partition else [],
+ options=settings,
+ ),
+ False,
+ )
+ table = catalog.get_table('default.events')
+ assert isinstance(table, FileStoreTable)
+ return table, schema
+
+
+def new_builder(table, fixed=False):
+ return (
+ table.new_postpone_fixed_bucket_write_builder()
+ if fixed
+ else table.new_batch_write_builder()
+ )
+
+
+def assert_backend(writer, native):
+ assert isinstance(writer, NativeTableWrite) == native
+ if native:
+ assert writer._python_writer is None
+ assert writer._native_writer is not None
+ assert type(writer._native_writer).__module__ ==
'pypaimon_rust.datafusion'
+
+
+def physical_rows(table, messages):
+ return [
+ row
+ for message in messages
+ for file in message.new_files
+ for row in pq.ParquetFile(file.file_path).read().to_pylist()
+ ]
+
+
+def read_merged_rows(table):
+ builder = table.new_read_builder()
+ splits = builder.new_scan().plan().splits()
+ assert splits and all(not split.raw_convertible for split in splits)
+ assert sum(len(split.files) for split in splits) == 2
+ return builder.new_read().to_arrow(splits).to_pylist()
+
+
[email protected](
+ 'bucket,options',
+ [
+ ('1', {}),
+ ('-1', {}),
+ ('-2', {}),
+ pytest.param('1', {'rowkind.field': None}, id='null-rowkind-field'),
+ ],
+)
+def test_object_delete_across_commits(tmp_path, native, bucket, options):
+ table, _ = make_table(
+ tmp_path, native, bucket, {'postpone.default-bucket-num': '1',
**options}
+ )
+ if 'rowkind.field' in options:
+ # Creation retains explicit nulls; table.copy() would remove this key.
+ assert 'rowkind.field' in table.options.options.to_map()
+ assert table.options.options.to_map()['rowkind.field'] is None
+ for kind in (RowKind.INSERT, RowKind.DELETE):
+ builder = new_builder(table, fixed=bucket == '-2')
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_row(GenericRow([1, 10], table.fields, kind))
+ assert_backend(writer, native)
+ messages = writer.prepare_commit()
+ assert [row['_VALUE_KIND'] for row in physical_rows(table,
messages)] == [
+ kind.value
+ ]
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ # Two overlapping files select merge reading, independently of
delete-count metadata.
+ assert read_merged_rows(table) == []
+
+
[email protected](
+ 'options,expected',
+ [
+ ({}, [0, 1, 2, 3]),
+ ({'rowkind.field': 'op'}, [0, 1, 2, 3]),
+ ({'ignore-delete': 'true'}, [0, 2]),
+ ({'ignore-update-before': 'true'}, [0, 2, 3]),
+ ({'ignore-delete': 'true', 'ignore-update-before': 'true'}, [0, 2]),
+ ({'rowkind.field': 'op', 'ignore-delete': 'true'}, [0, 2]),
+ ],
+)
+def test_object_events_and_configured_precedence(tmp_path, native, options,
expected):
+ # Ordinary postpone files preserve source events without merging equal
keys.
+ table, _ = make_table(tmp_path, native, '-2', options)
+ writer = new_builder(table).new_write()
+ try:
+ for kind in RowKind:
+ values: List[Any] = [kind.value, 10]
+ object_kind = kind
+ if 'rowkind.field' in options:
+ values.append(kind.to_string())
+ object_kind = RowKind((kind.value + 1) % 4)
+ writer.write_row(GenericRow(values, table.fields, object_kind))
+ assert_backend(writer, native)
+ rows = physical_rows(table, writer.prepare_commit())
+ assert [(row['id'], row['_VALUE_KIND']) for row in rows] == [
+ (k, k) for k in expected
+ ]
+ assert all(row['v'] == 10 for row in rows)
+ finally:
+ writer.close()
+
+
[email protected](
+ 'bucket,fixed,options',
+ [
+ ('1', False, {'ignore-delete': 'true'}),
+ ('-1', False, {'ignore-update-before': 'true'}),
+ ('-2', True, {'ignore-delete': 'true'}),
+ ],
+)
+def test_filter_precedes_invalid_routing(tmp_path, native, bucket, fixed,
options):
+ # Python's automatic postpone planner must also filter before sizing
inputs.
+ if native and fixed:
+ options = dict(options, **{'postpone.default-bucket-num': '1'})
+ table, _ = make_table(tmp_path, native, bucket, options, partition=True)
+ writer = new_builder(table, fixed).new_write()
+ try:
+ kinds = (
+ (RowKind.UPDATE_BEFORE, RowKind.DELETE)
+ if 'ignore-delete' in options
+ else (RowKind.UPDATE_BEFORE,)
+ )
+ for kind in kinds:
+ # These values cannot be hashed, converted to Arrow, or used as a
partition.
+ writer.write_row(GenericRow([[], object(), []], table.fields,
kind))
+ assert_backend(writer, native)
+ assert writer.prepare_commit() == []
+ assert not list(tmp_path.rglob('*.parquet'))
+ finally:
+ writer.close()
+
+
+def test_internal_row_events_and_arrow_default_insert(tmp_path, native):
+ table, schema = make_table(tmp_path, native)
+ builder = new_builder(table)
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([{'id': 10, 'v': 10}],
schema=schema))
+ for kind in RowKind:
+ writer.write_row(
+ ProjectedRow([0, 1]).replace_row(
+ GenericRow([kind.value, 20], table.fields, kind)
+ )
+ )
+ writer.write_row(GenericRow([11, 30], table.fields))
+ writer.write_arrow(pa.Table.from_pylist([{'id': 12, 'v': 40}],
schema=schema))
+ assert_backend(writer, native)
+ messages = writer.prepare_commit()
+ rows = sorted(physical_rows(table, messages), key=lambda row:
row['id'])
+ assert [(row['id'], row['_VALUE_KIND']) for row in rows] == [
+ (0, 0),
+ (1, 1),
+ (2, 2),
+ (3, 3),
+ (10, 0),
+ (11, 0),
+ (12, 0),
+ ]
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+
+
+def test_stream_object_events(tmp_path, native):
+ table, _ = make_table(tmp_path, native, '-2')
+ writer = table.new_stream_write_builder().new_write()
+ try:
+ for checkpoint, kind in enumerate(RowKind):
+ writer.write_row(GenericRow([checkpoint, 10], table.fields, kind))
+ assert_backend(writer, native)
+ rows = physical_rows(table, writer.prepare_commit(checkpoint))
+ assert [row['_VALUE_KIND'] for row in rows] == [kind.value]
+ finally:
+ writer.close()
+
+
+def test_arrow_filter_preserves_sliced_chunk_order_and_null_values(tmp_path,
native):
+ table, schema = make_table(
+ tmp_path,
+ native,
+ '-2',
+ {
+ 'rowkind.field': 'op',
+ 'ignore-delete': 'true',
+ 'ignore-update-before': 'true',
+ },
+ )
+ # The slice excludes two valid rows and spans chunks with repeated keys.
+ data = pa.table(
+ {
+ 'id': [99, 7, 2, 7, 4, 2, 98],
+ 'v': [99, 10, 20, None, 40, None, 98],
+ 'op': ['+I', '+I', '-D', '+U', '-U', '+I', '+I'],
+ },
+ schema=schema,
+ )
+ data = pa.Table.from_batches(data.to_batches(max_chunksize=4)).slice(1, 5)
+ writer = new_builder(table).new_write()
+ try:
+ writer.write_arrow(data)
+ assert_backend(writer, native)
+ messages = writer.prepare_commit()
+ rows = physical_rows(table, messages)
+ assert [(row['id'], row['v'], row['_VALUE_KIND']) for row in rows] == [
+ (7, 10, 0),
+ (7, None, 2),
+ (2, None, 0),
+ ]
+ assert [row['_SEQUENCE_NUMBER'] for row in rows] == [0, 1, 2]
+ assert all(
+ file.delete_row_count == 0 for msg in messages for file in
msg.new_files
+ )
+ finally:
+ writer.close()
+
+
+def test_empty_and_filtered_batches_do_not_affect_next_checkpoint(tmp_path,
native):
+ table, schema = make_table(
+ tmp_path,
+ native,
+ '-2',
+ {'rowkind.field': 'op', 'ignore-delete': 'true'},
+ )
+ writer = table.new_stream_write_builder().new_write()
+ try:
+ writer.write_arrow_batch(pa.RecordBatch.from_pylist([], schema=schema))
+ writer.write_arrow_batch(
+ pa.RecordBatch.from_pylist(
+ [
+ {'id': 1, 'v': 10, 'op': '-U'},
+ {'id': 2, 'v': 20, 'op': '-D'},
+ ],
+ schema=schema,
+ )
+ )
+ assert_backend(writer, native)
+ assert writer.prepare_commit(0) == []
+ assert not list(tmp_path.rglob('*.parquet'))
+
+ writer.write_arrow_batch(
+ pa.RecordBatch.from_pylist(
+ [{'id': 3, 'v': None, 'op': '+I'}], schema=schema
+ )
+ )
+ rows = physical_rows(table, writer.prepare_commit(1))
+ assert [
+ (row['id'], row['v'], row['_VALUE_KIND'], row['_SEQUENCE_NUMBER'])
+ for row in rows
+ ] == [(3, None, 0, 0)]
+ finally:
+ writer.close()
+
+
[email protected](
+ 'events,error',
+ [
+ (['+I', None], 'Unknown row kind string: None'),
+ (['-D', 'invalid'], 'Unknown row kind string: invalid'),
+ (['invalid', None], 'Unknown row kind string: invalid'),
+ ([None, 'invalid'], 'Unknown row kind string: None'),
+ ],
+)
+def test_invalid_arrow_events_reject_the_whole_batch(tmp_path, events, error):
+ table, schema = make_table(
+ tmp_path,
+ False,
+ '-2',
+ {'rowkind.field': 'op', 'ignore-delete': 'true'},
+ )
+ writer = new_builder(table).new_write()
+ try:
+ data = pa.record_batch(
+ {'id': [1, 2], 'v': [10, 20], 'op': events}, schema=schema
+ )
+ with pytest.raises(ValueError, match=error):
+ writer.write_arrow_batch(data)
+ assert writer.prepare_commit() == []
+ assert not list(tmp_path.rglob('*.parquet'))
+ finally:
+ writer.close()
diff --git a/paimon-python/pypaimon/tests/write/postpone_event_write_test.py
b/paimon-python/pypaimon/tests/write/postpone_event_write_test.py
index 16285e2a11..7664ee8792 100644
--- a/paimon-python/pypaimon/tests/write/postpone_event_write_test.py
+++ b/paimon-python/pypaimon/tests/write/postpone_event_write_test.py
@@ -58,11 +58,15 @@ def
test_postpone_preserves_unsorted_duplicate_rows(tmp_path, engine):
@pytest.mark.parametrize('row_input', [False, True])
[email protected]('options,expected', [
- ({}, [0, 1, 2, 3]),
- ({'ignore-delete': 'true'}, [0, 2]),
- ({'ignore-update-before': 'true'}, [0, 2, 3]),
-])
[email protected](
+ 'options,expected',
+ [
+ ({}, [0, 1, 2, 3]),
+ ({'ignore-delete': 'true'}, [0, 2]),
+ ({'ignore-update-before': 'true'}, [0, 2, 3]),
+ ({'ignore-delete': 'true', 'ignore-update-before': 'true'}, [0, 2]),
+ ],
+)
def test_python_postpone_row_kinds_are_filtered_before_writing(tmp_path,
row_input, options, expected):
from pypaimon.table.row.generic_row import GenericRow
table, schema = make_table(tmp_path, {'rowkind.field': 'op', **options})
diff --git a/paimon-python/pypaimon/write/postpone_batch_table_write.py
b/paimon-python/pypaimon/write/postpone_batch_table_write.py
index f00f15b544..f15f0bafb5 100644
--- a/paimon-python/pypaimon/write/postpone_batch_table_write.py
+++ b/paimon-python/pypaimon/write/postpone_batch_table_write.py
@@ -156,6 +156,10 @@ class PostponeFixedBucketBatchTableWrite(BatchTableWrite):
)
require_columns(values_by_name, column_names, "write_row")
require_columns(values_by_name, self.table.partition_keys, "write_row")
+ from pypaimon.write.row_kind import skip_write_row
+
+ if skip_write_row(self.table, values_by_name, row.get_row_kind()):
+ return
partition = tuple(
values_by_name[key] for key in self.table.partition_keys)
if self._bucket_plan.contains(partition):
diff --git a/paimon-python/pypaimon/write/row_kind.py
b/paimon-python/pypaimon/write/row_kind.py
index 78b5bc5e92..df279aadf6 100644
--- a/paimon-python/pypaimon/write/row_kind.py
+++ b/paimon-python/pypaimon/write/row_kind.py
@@ -16,23 +16,39 @@
"""Configured row kinds and pre-routing filters, matching Java
RowKindFilter."""
+from typing import List, Union
+
import pyarrow as pa
+import pyarrow.compute as pc
+from pypaimon.common.options.core_options import CoreOptions
from pypaimon.table.row.row_kind import RowKind
+from pypaimon.table.special_fields import SpecialFields
-def row_kinds(options, data):
- name = options.options.to_map().get('rowkind.field')
- if name is None:
- if '_VALUE_KIND' in data.schema.names:
- return data.column('_VALUE_KIND').to_pylist()
- return [0] * data.num_rows
- if name not in data.schema.names:
- raise ValueError('Cannot find rowkind field %s in table schema' % name)
- column = data.column(name)
+def row_kinds(
+ options: CoreOptions,
+ data: Union[pa.Table, pa.RecordBatch],
+) -> Union[pa.Array, pa.ChunkedArray]:
+ """Resolve events as an int8 Arrow column, reusing existing event data."""
+ row_kind_field_name = options.options.to_map().get('rowkind.field')
+ if row_kind_field_name is None:
+ if SpecialFields.VALUE_KIND.name in data.schema.names:
+ return data.column(SpecialFields.VALUE_KIND.name)
+ return pa.repeat(pa.scalar(RowKind.INSERT.value, type=pa.int8()),
data.num_rows)
+ if row_kind_field_name not in data.schema.names:
+ raise ValueError(
+ 'Cannot find rowkind field %s in table schema' %
row_kind_field_name
+ )
+ column = data.column(row_kind_field_name)
if not pa.types.is_string(column.type):
raise ValueError('rowkind.field column must be a string')
- return [RowKind.from_string(value).value for value in column.to_pylist()]
+ kinds: List[int] = []
+ for value in column.to_pylist():
+ if value is None:
+ raise ValueError('Unknown row kind string: None')
+ kinds.append(RowKind.from_string(value).value)
+ return pa.array(kinds, type=pa.int8())
def filter_write_batch(table, data):
@@ -47,8 +63,12 @@ def filter_write_batch(table, data):
ignore_before = str(raw.get('ignore-update-before', 'false')).lower() ==
'true'
if not ignore_delete and not ignore_before:
return data
- keep = [not _is_filtered(kind, ignore_delete, ignore_before) for kind in
kinds]
- return data.filter(pa.array(keep, type=pa.bool_()))
+ ignored = [RowKind.UPDATE_BEFORE.value]
+ if ignore_delete:
+ ignored.append(RowKind.DELETE.value)
+ return data.filter(
+ pc.invert(pc.is_in(kinds, value_set=pa.array(ignored, type=pa.int8())))
+ )
def with_row_kind(table, data, row):
@@ -57,10 +77,15 @@ def with_row_kind(table, data, row):
A configured rowkind.field takes precedence, as in Java RowKindGenerator.
Public Arrow writes still accept only the declared table/write schema.
"""
- if table.is_primary_key_table and
table.options.options.to_map().get('rowkind.field') is None:
- return data.append_column(
- pa.field('_VALUE_KIND', pa.int8(), nullable=False),
- pa.array([row.get_row_kind().value] * data.num_rows,
type=pa.int8()))
+ if (
+ table.is_primary_key_table
+ and table.options.options.to_map().get('rowkind.field') is None
+ ):
+ kind = row.get_row_kind()
+ data = data.append_column(
+ pa.field(SpecialFields.VALUE_KIND.name, pa.int8(), nullable=False),
+ pa.repeat(pa.scalar(kind.value, type=pa.int8()), data.num_rows),
+ )
return data
@@ -69,8 +94,12 @@ def skip_write_row(table, values, row_kind=RowKind.INSERT):
return False
options = table.options
raw = options.options.to_map()
- name = raw.get('rowkind.field')
- kind = row_kind.value if name is None else
RowKind.from_string(values[name]).value
+ row_kind_field_name = raw.get('rowkind.field')
+ kind = (
+ row_kind.value
+ if row_kind_field_name is None
+ else RowKind.from_string(values[row_kind_field_name]).value
+ )
return _is_filtered(kind, options.ignore_delete(),
str(raw.get('ignore-update-before', 'false')).lower()
== 'true')
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index d6ded1e0f0..98cdd66d2b 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -30,6 +30,7 @@ from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.table.bucket_mode import BucketMode
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.row_kind import RowKind
+from pypaimon.table.special_fields import SpecialFields
from pypaimon.utils.file_store_path_factory import canonical_data_file_path
from pypaimon.write.map_shared_shredding_writer import MapSharedShreddingWriter
from pypaimon.write.writer.mosaic_writer_options import
create_mosaic_writer_options
@@ -432,9 +433,12 @@ class DataWriter(ABC):
def _count_delete_rows(self, data: pa.Table) -> int:
# Count the final file contents, after merging and rolling the buffer.
- if not self.table.is_primary_key_table or '_VALUE_KIND' not in
data.schema.names:
+ if (
+ not self.table.is_primary_key_table
+ or SpecialFields.VALUE_KIND.name not in data.schema.names
+ ):
return 0
- kind_counts = pc.value_counts(data.column('_VALUE_KIND'))
+ kind_counts =
pc.value_counts(data.column(SpecialFields.VALUE_KIND.name))
return sum(entry['counts'].as_py() for entry in kind_counts
if entry['values'].as_py() in (RowKind.UPDATE_BEFORE.value,
RowKind.DELETE.value))
diff --git a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
index b2e79d50d0..f513e148d6 100644
--- a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
@@ -26,6 +26,7 @@ from pypaimon.read.reader.deduplicate_merge_function import \
DeduplicateMergeFunction
from pypaimon.common.options.core_options import ChangelogProducer
from pypaimon.table.row.key_value import KeyValue
+from pypaimon.table.special_fields import SpecialFields
from pypaimon.write.writer.data_writer import DataWriter
from pypaimon.write.writer.write_buffer import WriteBuffer
@@ -313,12 +314,15 @@ class KeyValueDataWriter(DataWriter):
new_fields.append(pa.field('_SEQUENCE_NUMBER', pa.int64(),
nullable=False))
from pypaimon.write.row_kind import row_kinds
- value_kind_column = pa.array(row_kinds(self.options, data),
type=pa.int8())
+
+ value_kind_column = row_kinds(self.options, data)
new_arrays.append(value_kind_column)
- new_fields.append(pa.field('_VALUE_KIND', pa.int8(), nullable=False))
+ new_fields.append(
+ pa.field(SpecialFields.VALUE_KIND.name, pa.int8(), nullable=False)
+ )
for i in range(data.num_columns):
- if data.schema.field(i).name == '_VALUE_KIND':
+ if data.schema.field(i).name == SpecialFields.VALUE_KIND.name:
continue
new_arrays.append(data.column(i))
new_fields.append(data.schema.field(i))
