Remove `numpy` as a dependency

[kevinjqliu]

Feature Request / Improvement

#1256 mentioned removing numpy as a dependency. numpy is currently only used in 1 function

iceberg-python/pyiceberg/io/pyarrow.py

Lines 810 to 815 in 583a7e9

	def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], start_index: int, end_index: int) -> pa.Array:
	if len(positional_deletes) == 1:
	all_chunks = positional_deletes[0]
	else:
	all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
	return np.subtract(np.setdiff1d(np.arange(start_index, end_index), all_chunks, assume_unique=False), start_index)

and 2 tests

iceberg-python/tests/integration/test_writes/test_writes.py

Lines 1348 to 1383 in 583a7e9

	def test_delete_threshold(session_catalog: Catalog) -> None:
	schema = Schema(
	NestedField(field_id=101, name="id", field_type=LongType(), required=True),
	NestedField(field_id=103, name="created_at", field_type=DateType(), required=False),
	NestedField(field_id=104, name="relevancy_score", field_type=DoubleType(), required=False),
	)
	partition_spec = PartitionSpec(PartitionField(source_id=103, field_id=2000, transform=DayTransform(), name="created_at_day"))
	try:
	session_catalog.drop_table(
	identifier="default.scores",
	)
	except NoSuchTableError:
	pass
	session_catalog.create_table(
	identifier="default.scores",
	schema=schema,
	partition_spec=partition_spec,
	)
	# Parameters
	num_rows = 100 # Number of rows in the dataframe
	id_min, id_max = 1, 10000
	date_start, date_end = date(2024, 1, 1), date(2024, 2, 1)
	# Generate the 'id' column
	id_column = np.random.randint(id_min, id_max, num_rows)
	# Generate the 'created_at' column as dates only
	date_range = pd.date_range(start=date_start, end=date_end, freq="D") # Daily frequency for dates
	created_at_column = np.random.choice(date_range, num_rows) # Convert to string (YYYY-MM-DD format)
	# Generate the 'relevancy_score' column with a peak around 0.1
	relevancy_score_column = np.random.beta(a=2, b=20, size=num_rows) # Adjusting parameters to peak around 0.1

iceberg-python/tests/integration/test_writes/test_writes.py

Lines 1404 to 1412 in 583a7e9

	@pytest.mark.integration
	def test_rewrite_manifest_after_partition_evolution(session_catalog: Catalog) -> None:
	np.random.seed(876)
	N = 1440
	d = {
	"timestamp": pa.array([datetime(2023, 1, 1, 0, 0, 0) + timedelta(minutes=i) for i in range(N)]),
	"category": pa.array([np.random.choice(["A", "B", "C"]) for _ in range(N)]),
	"value": pa.array(np.random.normal(size=N)),
	}

It seems like we can replace the _combine_positional_deletes use of numpy with pure python operators or pyarrow operators

[kevinjqliu]

This ChatGPT-generated function seems to be logically equivalent. And pass integration tests.
But we should double check this

def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], start_index: int, end_index: int) -> pa.Array:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 
 # Create the full range array with pyarrow
 full_range = pa.array(range(start_index, end_index))
 
 # Filter out values in all_chunks from full_range
 result = pc.filter(full_range, pc.invert(pc.is_in(full_range, value_set=all_chunks)))
 
 # Subtract the start_index from each element in the result
 adjusted_result = pc.subtract(result, pa.scalar(start_index))
 
 return adjusted_result

[Fokko]

@kevinjqliu That looks about right to me :) Could you do some benchmarking to see if we don't get too much of a performance regression?

[Fokko]

I took the liberty of checking this, so we can fix this for 0.8.0:

It looks like we're spending a lot of time on generating the sequence. I think it would be good to add this to Arrow itself: apache/arrow#44583 I think we can move this forward since it doesn't happy very often that a file is affected by multiple positional deletes. Once it has been added to Arrow, we can also add it to PyIceberg

[corleyma]

It looks like we're spending a lot of time on generating the sequence. I think it would be good to add this to Arrow itself: apache/arrow#44583 I think we can move this forward since it doesn't happy very often that a file is affected by multiple positional deletes. Once it has been added to Arrow, we can also add it to PyIceberg

@Fokko
If we wanted, I suspect we could use the pyarrow cython/C++ API to allocate and create the range array much faster.

Something like:

# range_impl.pyx
# distutils: language=c++

from pyarrow.lib cimport *
import pyarrow as pa
import pyarrow.compute as pc
from libcpp.memory cimport shared_ptr, make_shared
from libcpp.vector cimport vector
from libc.stdint cimport int64_t, uint8_t

def create_arrow_range(int64_t start, int64_t end):
 if start >= end:
 raise ValueError("start must be less than end")
 
 cdef:
 vector[int64_t] values
 int64_t i
 shared_ptr[CArray] out
 shared_ptr[CBuffer] buffer
 shared_ptr[CDataType] dtype
 
 # Pre-allocate the vector size
 values.reserve(end - start)
 
 # Fill vector with range values
 for i in range(start, end):
 values.push_back(i)
 
 # Create buffer from vector data
 buffer = make_shared[CBuffer](
 <uint8_t*>values.data(),
 values.size() * sizeof(int64_t)
 )
 
 # Get int64 datatype
 dtype = pyarrow_unwrap_data_type(pa.int64())
 
 # Create Int64Array
 out = make_shared[CInt64Array](dtype, end - start, buffer)
 
 # Wrap and return as Python object
 return pyarrow_wrap_array(out)
 

# setup.py
import os

import numpy as np
import pyarrow as pa
from Cython.Build import cythonize
from setuptools import Extension, setup

# Ensure you have the Arrow C++ shared libraries available
try:
 pa.create_library_symlinks()
except:
 pass

ext_modules = cythonize(
 "range_impl.pyx",
 compiler_directives={
 "language_level": "3",
 "embedsignature": True,
 },
)

for ext in ext_modules:
 # We only need numpy headers at build time
 ext.include_dirs.append(np.get_include())
 ext.include_dirs.append(pa.get_include())
 ext.libraries.extend(pa.get_libraries())
 ext.library_dirs.extend(pa.get_library_dirs())

 # Enable C++17
 if os.name == "posix":
 ext.extra_compile_args.append("-std=c++17")

setup(
 name="arrow_range",
 ext_modules=ext_modules,
 setup_requires=["pyarrow", "numpy"], # Only needed during build
 install_requires=["pyarrow"], # Runtime dependency
)

and using this to benchmark:

# benchmark.py
import time

import pyarrow as pa

from range_impl import create_arrow_range


def test_python_range(start, end):
 return pa.array(range(start, end))


def test_cython_range(start, end):
 return create_arrow_range(start, end)


def benchmark(start, end, n_runs=5):
 times_python = []
 times_cython = []

 print(f"Testing range from {start} to {end} ({end-start} elements)")

 # Warm up
 _ = test_python_range(0, 1000)
 _ = test_cython_range(0, 1000)

 for i in range(n_runs):
 # Test Python range
 t0 = time.perf_counter()
 arr_py = test_python_range(start, end)
 t1 = time.perf_counter()
 times_python.append(t1 - t0)

 # Test Cython range
 t0 = time.perf_counter()
 arr_cy = test_cython_range(start, end)
 t1 = time.perf_counter()
 times_cython.append(t1 - t0)

 # Verify results match
 assert arr_py.equals(arr_cy), "Arrays don't match!"

 avg_python = sum(times_python) / len(times_python)
 avg_cython = sum(times_cython) / len(times_cython)

 print(f"Python average: {avg_python:.4f}s")
 print(f"Cython average: {avg_cython:.4f}s")
 print(f"Speedup: {avg_python/avg_cython:.2f}x")


if __name__ == "__main__":
 benchmark(0, 10_000_000) 

Gives me this on my m1:

python setup.py build_ext --inplace
python benchmark.py
running build_ext
Testing range from 0 to 10000000 (10000000 elements)
Python average: 0.9227s
Cython average: 0.0125s
Speedup: 73.64x

[kevinjqliu]

supercool! As a project, I don't want to maintain an implementation of the range function 😄 Is there a way to contribute create_arrow_range to Arrow?

[corleyma]

Arrow's release cycles are... not quick, but I have no doubt that logic will land there at some point.