1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
|
# 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.
# cython: language_level = 3
import sys
from cython.operator cimport dereference as deref
from collections import namedtuple
from pyarrow.lib import frombytes, tobytes, ordered_dict
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
import pyarrow.lib as lib
import numpy as np
cdef wrap_scalar_function(const shared_ptr[CFunction]& sp_func):
"""
Wrap a C++ scalar Function in a ScalarFunction object.
"""
cdef ScalarFunction func = ScalarFunction.__new__(ScalarFunction)
func.init(sp_func)
return func
cdef wrap_vector_function(const shared_ptr[CFunction]& sp_func):
"""
Wrap a C++ vector Function in a VectorFunction object.
"""
cdef VectorFunction func = VectorFunction.__new__(VectorFunction)
func.init(sp_func)
return func
cdef wrap_scalar_aggregate_function(const shared_ptr[CFunction]& sp_func):
"""
Wrap a C++ aggregate Function in a ScalarAggregateFunction object.
"""
cdef ScalarAggregateFunction func = \
ScalarAggregateFunction.__new__(ScalarAggregateFunction)
func.init(sp_func)
return func
cdef wrap_hash_aggregate_function(const shared_ptr[CFunction]& sp_func):
"""
Wrap a C++ aggregate Function in a HashAggregateFunction object.
"""
cdef HashAggregateFunction func = \
HashAggregateFunction.__new__(HashAggregateFunction)
func.init(sp_func)
return func
cdef wrap_meta_function(const shared_ptr[CFunction]& sp_func):
"""
Wrap a C++ meta Function in a MetaFunction object.
"""
cdef MetaFunction func = MetaFunction.__new__(MetaFunction)
func.init(sp_func)
return func
cdef wrap_function(const shared_ptr[CFunction]& sp_func):
"""
Wrap a C++ Function in a Function object.
This dispatches to specialized wrappers depending on the function kind.
"""
if sp_func.get() == NULL:
raise ValueError("Function was NULL")
cdef FunctionKind c_kind = sp_func.get().kind()
if c_kind == FunctionKind_SCALAR:
return wrap_scalar_function(sp_func)
elif c_kind == FunctionKind_VECTOR:
return wrap_vector_function(sp_func)
elif c_kind == FunctionKind_SCALAR_AGGREGATE:
return wrap_scalar_aggregate_function(sp_func)
elif c_kind == FunctionKind_HASH_AGGREGATE:
return wrap_hash_aggregate_function(sp_func)
elif c_kind == FunctionKind_META:
return wrap_meta_function(sp_func)
else:
raise NotImplementedError("Unknown Function::Kind")
cdef wrap_scalar_kernel(const CScalarKernel* c_kernel):
if c_kernel == NULL:
raise ValueError("Kernel was NULL")
cdef ScalarKernel kernel = ScalarKernel.__new__(ScalarKernel)
kernel.init(c_kernel)
return kernel
cdef wrap_vector_kernel(const CVectorKernel* c_kernel):
if c_kernel == NULL:
raise ValueError("Kernel was NULL")
cdef VectorKernel kernel = VectorKernel.__new__(VectorKernel)
kernel.init(c_kernel)
return kernel
cdef wrap_scalar_aggregate_kernel(const CScalarAggregateKernel* c_kernel):
if c_kernel == NULL:
raise ValueError("Kernel was NULL")
cdef ScalarAggregateKernel kernel = \
ScalarAggregateKernel.__new__(ScalarAggregateKernel)
kernel.init(c_kernel)
return kernel
cdef wrap_hash_aggregate_kernel(const CHashAggregateKernel* c_kernel):
if c_kernel == NULL:
raise ValueError("Kernel was NULL")
cdef HashAggregateKernel kernel = \
HashAggregateKernel.__new__(HashAggregateKernel)
kernel.init(c_kernel)
return kernel
cdef class Kernel(_Weakrefable):
"""
A kernel object.
Kernels handle the execution of a Function for a certain signature.
"""
def __init__(self):
raise TypeError("Do not call {}'s constructor directly"
.format(self.__class__.__name__))
cdef class ScalarKernel(Kernel):
cdef const CScalarKernel* kernel
cdef void init(self, const CScalarKernel* kernel) except *:
self.kernel = kernel
def __repr__(self):
return ("ScalarKernel<{}>"
.format(frombytes(self.kernel.signature.get().ToString())))
cdef class VectorKernel(Kernel):
cdef const CVectorKernel* kernel
cdef void init(self, const CVectorKernel* kernel) except *:
self.kernel = kernel
def __repr__(self):
return ("VectorKernel<{}>"
.format(frombytes(self.kernel.signature.get().ToString())))
cdef class ScalarAggregateKernel(Kernel):
cdef const CScalarAggregateKernel* kernel
cdef void init(self, const CScalarAggregateKernel* kernel) except *:
self.kernel = kernel
def __repr__(self):
return ("ScalarAggregateKernel<{}>"
.format(frombytes(self.kernel.signature.get().ToString())))
cdef class HashAggregateKernel(Kernel):
cdef const CHashAggregateKernel* kernel
cdef void init(self, const CHashAggregateKernel* kernel) except *:
self.kernel = kernel
def __repr__(self):
return ("HashAggregateKernel<{}>"
.format(frombytes(self.kernel.signature.get().ToString())))
FunctionDoc = namedtuple(
"FunctionDoc",
("summary", "description", "arg_names", "options_class"))
cdef class Function(_Weakrefable):
"""
A compute function.
A function implements a certain logical computation over a range of
possible input signatures. Each signature accepts a range of input
types and is implemented by a given Kernel.
Functions can be of different kinds:
* "scalar" functions apply an item-wise computation over all items
of their inputs. Each item in the output only depends on the values
of the inputs at the same position. Examples: addition, comparisons,
string predicates...
* "vector" functions apply a collection-wise computation, such that
each item in the output may depend on the values of several items
in each input. Examples: dictionary encoding, sorting, extracting
unique values...
* "scalar_aggregate" functions reduce the dimensionality of the inputs by
applying a reduction function. Examples: sum, min_max, mode...
* "hash_aggregate" functions apply a reduction function to an input
subdivided by grouping criteria. They may not be directly called.
Examples: hash_sum, hash_min_max...
* "meta" functions dispatch to other functions.
"""
cdef:
shared_ptr[CFunction] sp_func
CFunction* base_func
_kind_map = {
FunctionKind_SCALAR: "scalar",
FunctionKind_VECTOR: "vector",
FunctionKind_SCALAR_AGGREGATE: "scalar_aggregate",
FunctionKind_HASH_AGGREGATE: "hash_aggregate",
FunctionKind_META: "meta",
}
def __init__(self):
raise TypeError("Do not call {}'s constructor directly"
.format(self.__class__.__name__))
cdef void init(self, const shared_ptr[CFunction]& sp_func) except *:
self.sp_func = sp_func
self.base_func = sp_func.get()
def __repr__(self):
return ("arrow.compute.Function<name={}, kind={}, "
"arity={}, num_kernels={}>"
.format(self.name, self.kind, self.arity, self.num_kernels))
def __reduce__(self):
# Reduction uses the global registry
return get_function, (self.name,)
@property
def name(self):
"""
The function name.
"""
return frombytes(self.base_func.name())
@property
def arity(self):
"""
The function arity.
If Ellipsis (i.e. `...`) is returned, the function takes a variable
number of arguments.
"""
cdef CArity arity = self.base_func.arity()
if arity.is_varargs:
return ...
else:
return arity.num_args
@property
def kind(self):
"""
The function kind.
"""
cdef FunctionKind c_kind = self.base_func.kind()
try:
return self._kind_map[c_kind]
except KeyError:
raise NotImplementedError("Unknown Function::Kind")
@property
def _doc(self):
"""
The C++-like function documentation (for internal use).
"""
cdef CFunctionDoc c_doc = self.base_func.doc()
return FunctionDoc(frombytes(c_doc.summary),
frombytes(c_doc.description),
[frombytes(s) for s in c_doc.arg_names],
frombytes(c_doc.options_class))
@property
def num_kernels(self):
"""
The number of kernels implementing this function.
"""
return self.base_func.num_kernels()
def call(self, args, FunctionOptions options=None,
MemoryPool memory_pool=None):
"""
Call the function on the given arguments.
"""
cdef:
const CFunctionOptions* c_options = NULL
CMemoryPool* pool = maybe_unbox_memory_pool(memory_pool)
CExecContext c_exec_ctx = CExecContext(pool)
vector[CDatum] c_args
CDatum result
_pack_compute_args(args, &c_args)
if options is not None:
c_options = options.get_options()
with nogil:
result = GetResultValue(
self.base_func.Execute(c_args, c_options, &c_exec_ctx)
)
return wrap_datum(result)
cdef class ScalarFunction(Function):
cdef const CScalarFunction* func
cdef void init(self, const shared_ptr[CFunction]& sp_func) except *:
Function.init(self, sp_func)
self.func = <const CScalarFunction*> sp_func.get()
@property
def kernels(self):
"""
The kernels implementing this function.
"""
cdef vector[const CScalarKernel*] kernels = self.func.kernels()
return [wrap_scalar_kernel(k) for k in kernels]
cdef class VectorFunction(Function):
cdef const CVectorFunction* func
cdef void init(self, const shared_ptr[CFunction]& sp_func) except *:
Function.init(self, sp_func)
self.func = <const CVectorFunction*> sp_func.get()
@property
def kernels(self):
"""
The kernels implementing this function.
"""
cdef vector[const CVectorKernel*] kernels = self.func.kernels()
return [wrap_vector_kernel(k) for k in kernels]
cdef class ScalarAggregateFunction(Function):
cdef const CScalarAggregateFunction* func
cdef void init(self, const shared_ptr[CFunction]& sp_func) except *:
Function.init(self, sp_func)
self.func = <const CScalarAggregateFunction*> sp_func.get()
@property
def kernels(self):
"""
The kernels implementing this function.
"""
cdef vector[const CScalarAggregateKernel*] kernels = \
self.func.kernels()
return [wrap_scalar_aggregate_kernel(k) for k in kernels]
cdef class HashAggregateFunction(Function):
cdef const CHashAggregateFunction* func
cdef void init(self, const shared_ptr[CFunction]& sp_func) except *:
Function.init(self, sp_func)
self.func = <const CHashAggregateFunction*> sp_func.get()
@property
def kernels(self):
"""
The kernels implementing this function.
"""
cdef vector[const CHashAggregateKernel*] kernels = self.func.kernels()
return [wrap_hash_aggregate_kernel(k) for k in kernels]
cdef class MetaFunction(Function):
cdef const CMetaFunction* func
cdef void init(self, const shared_ptr[CFunction]& sp_func) except *:
Function.init(self, sp_func)
self.func = <const CMetaFunction*> sp_func.get()
# Since num_kernels is exposed, also expose a kernels property
@property
def kernels(self):
"""
The kernels implementing this function.
"""
return []
cdef _pack_compute_args(object values, vector[CDatum]* out):
for val in values:
if isinstance(val, (list, np.ndarray)):
val = lib.asarray(val)
if isinstance(val, Array):
out.push_back(CDatum((<Array> val).sp_array))
continue
elif isinstance(val, ChunkedArray):
out.push_back(CDatum((<ChunkedArray> val).sp_chunked_array))
continue
elif isinstance(val, Scalar):
out.push_back(CDatum((<Scalar> val).unwrap()))
continue
elif isinstance(val, RecordBatch):
out.push_back(CDatum((<RecordBatch> val).sp_batch))
continue
elif isinstance(val, Table):
out.push_back(CDatum((<Table> val).sp_table))
continue
else:
# Is it a Python scalar?
try:
scal = lib.scalar(val)
except Exception:
# Raise dedicated error below
pass
else:
out.push_back(CDatum((<Scalar> scal).unwrap()))
continue
raise TypeError(f"Got unexpected argument type {type(val)} "
"for compute function")
cdef class FunctionRegistry(_Weakrefable):
cdef CFunctionRegistry* registry
def __init__(self):
self.registry = GetFunctionRegistry()
def list_functions(self):
"""
Return all function names in the registry.
"""
cdef vector[c_string] names = self.registry.GetFunctionNames()
return [frombytes(name) for name in names]
def get_function(self, name):
"""
Look up a function by name in the registry.
Parameters
----------
name : str
The name of the function to lookup
"""
cdef:
c_string c_name = tobytes(name)
shared_ptr[CFunction] func
with nogil:
func = GetResultValue(self.registry.GetFunction(c_name))
return wrap_function(func)
cdef FunctionRegistry _global_func_registry = FunctionRegistry()
def function_registry():
return _global_func_registry
def get_function(name):
"""
Get a function by name.
The function is looked up in the global registry
(as returned by `function_registry()`).
Parameters
----------
name : str
The name of the function to lookup
"""
return _global_func_registry.get_function(name)
def list_functions():
"""
Return all function names in the global registry.
"""
return _global_func_registry.list_functions()
def call_function(name, args, options=None, memory_pool=None):
"""
Call a named function.
The function is looked up in the global registry
(as returned by `function_registry()`).
Parameters
----------
name : str
The name of the function to call.
args : list
The arguments to the function.
options : optional
options provided to the function.
memory_pool : MemoryPool, optional
memory pool to use for allocations during function execution.
"""
func = _global_func_registry.get_function(name)
return func.call(args, options=options, memory_pool=memory_pool)
cdef class FunctionOptions(_Weakrefable):
__slots__ = () # avoid mistakingly creating attributes
cdef const CFunctionOptions* get_options(self) except NULL:
return self.wrapped.get()
cdef void init(self, unique_ptr[CFunctionOptions] options):
self.wrapped = move(options)
def serialize(self):
cdef:
CResult[shared_ptr[CBuffer]] res = self.get_options().Serialize()
shared_ptr[CBuffer] c_buf = GetResultValue(res)
return pyarrow_wrap_buffer(c_buf)
@staticmethod
def deserialize(buf):
"""
Deserialize options for a function.
Parameters
----------
buf : Buffer
The buffer containing the data to deserialize.
"""
cdef:
shared_ptr[CBuffer] c_buf = pyarrow_unwrap_buffer(buf)
CResult[unique_ptr[CFunctionOptions]] maybe_options = \
DeserializeFunctionOptions(deref(c_buf))
unique_ptr[CFunctionOptions] c_options
c_options = move(GetResultValue(move(maybe_options)))
type_name = frombytes(c_options.get().options_type().type_name())
module = globals()
if type_name not in module:
raise ValueError(f'Cannot deserialize "{type_name}"')
klass = module[type_name]
options = klass.__new__(klass)
(<FunctionOptions> options).init(move(c_options))
return options
def __repr__(self):
type_name = self.__class__.__name__
# Remove {} so we can use our own braces
string_repr = frombytes(self.get_options().ToString())[1:-1]
return f"{type_name}({string_repr})"
def __eq__(self, FunctionOptions other):
return self.get_options().Equals(deref(other.get_options()))
def _raise_invalid_function_option(value, description, *,
exception_class=ValueError):
raise exception_class(f"\"{value}\" is not a valid {description}")
# NOTE:
# To properly expose the constructor signature of FunctionOptions
# subclasses, we use a two-level inheritance:
# 1. a C extension class that implements option validation and setting
# (won't expose function signatures because of
# https://github.com/cython/cython/issues/3873)
# 2. a Python derived class that implements the constructor
cdef class _CastOptions(FunctionOptions):
cdef CCastOptions* options
cdef void init(self, unique_ptr[CFunctionOptions] options):
FunctionOptions.init(self, move(options))
self.options = <CCastOptions*> self.wrapped.get()
def _set_options(self, DataType target_type, allow_int_overflow,
allow_time_truncate, allow_time_overflow,
allow_decimal_truncate, allow_float_truncate,
allow_invalid_utf8):
self.init(unique_ptr[CFunctionOptions](new CCastOptions()))
self._set_type(target_type)
if allow_int_overflow is not None:
self.allow_int_overflow = allow_int_overflow
if allow_time_truncate is not None:
self.allow_time_truncate = allow_time_truncate
if allow_time_overflow is not None:
self.allow_time_overflow = allow_time_overflow
if allow_decimal_truncate is not None:
self.allow_decimal_truncate = allow_decimal_truncate
if allow_float_truncate is not None:
self.allow_float_truncate = allow_float_truncate
if allow_invalid_utf8 is not None:
self.allow_invalid_utf8 = allow_invalid_utf8
def _set_type(self, target_type=None):
if target_type is not None:
deref(self.options).to_type = \
(<DataType> ensure_type(target_type)).sp_type
def _set_safe(self):
self.init(unique_ptr[CFunctionOptions](
new CCastOptions(CCastOptions.Safe())))
def _set_unsafe(self):
self.init(unique_ptr[CFunctionOptions](
new CCastOptions(CCastOptions.Unsafe())))
def is_safe(self):
return not (deref(self.options).allow_int_overflow or
deref(self.options).allow_time_truncate or
deref(self.options).allow_time_overflow or
deref(self.options).allow_decimal_truncate or
deref(self.options).allow_float_truncate or
deref(self.options).allow_invalid_utf8)
@property
def allow_int_overflow(self):
return deref(self.options).allow_int_overflow
@allow_int_overflow.setter
def allow_int_overflow(self, c_bool flag):
deref(self.options).allow_int_overflow = flag
@property
def allow_time_truncate(self):
return deref(self.options).allow_time_truncate
@allow_time_truncate.setter
def allow_time_truncate(self, c_bool flag):
deref(self.options).allow_time_truncate = flag
@property
def allow_time_overflow(self):
return deref(self.options).allow_time_overflow
@allow_time_overflow.setter
def allow_time_overflow(self, c_bool flag):
deref(self.options).allow_time_overflow = flag
@property
def allow_decimal_truncate(self):
return deref(self.options).allow_decimal_truncate
@allow_decimal_truncate.setter
def allow_decimal_truncate(self, c_bool flag):
deref(self.options).allow_decimal_truncate = flag
@property
def allow_float_truncate(self):
return deref(self.options).allow_float_truncate
@allow_float_truncate.setter
def allow_float_truncate(self, c_bool flag):
deref(self.options).allow_float_truncate = flag
@property
def allow_invalid_utf8(self):
return deref(self.options).allow_invalid_utf8
@allow_invalid_utf8.setter
def allow_invalid_utf8(self, c_bool flag):
deref(self.options).allow_invalid_utf8 = flag
class CastOptions(_CastOptions):
def __init__(self, target_type=None, *, allow_int_overflow=None,
allow_time_truncate=None, allow_time_overflow=None,
allow_decimal_truncate=None, allow_float_truncate=None,
allow_invalid_utf8=None):
self._set_options(target_type, allow_int_overflow, allow_time_truncate,
allow_time_overflow, allow_decimal_truncate,
allow_float_truncate, allow_invalid_utf8)
@staticmethod
def safe(target_type=None):
""""
Create a CastOptions for a safe cast.
Parameters
----------
target_type : optional
Target cast type for the safe cast.
"""
self = CastOptions()
self._set_safe()
self._set_type(target_type)
return self
@staticmethod
def unsafe(target_type=None):
""""
Create a CastOptions for an unsafe cast.
Parameters
----------
target_type : optional
Target cast type for the unsafe cast.
"""
self = CastOptions()
self._set_unsafe()
self._set_type(target_type)
return self
cdef class _ElementWiseAggregateOptions(FunctionOptions):
def _set_options(self, skip_nulls):
self.wrapped.reset(new CElementWiseAggregateOptions(skip_nulls))
class ElementWiseAggregateOptions(_ElementWiseAggregateOptions):
def __init__(self, *, skip_nulls=True):
self._set_options(skip_nulls)
cdef CRoundMode unwrap_round_mode(round_mode) except *:
if round_mode == "down":
return CRoundMode_DOWN
elif round_mode == "up":
return CRoundMode_UP
elif round_mode == "towards_zero":
return CRoundMode_TOWARDS_ZERO
elif round_mode == "towards_infinity":
return CRoundMode_TOWARDS_INFINITY
elif round_mode == "half_down":
return CRoundMode_HALF_DOWN
elif round_mode == "half_up":
return CRoundMode_HALF_UP
elif round_mode == "half_towards_zero":
return CRoundMode_HALF_TOWARDS_ZERO
elif round_mode == "half_towards_infinity":
return CRoundMode_HALF_TOWARDS_INFINITY
elif round_mode == "half_to_even":
return CRoundMode_HALF_TO_EVEN
elif round_mode == "half_to_odd":
return CRoundMode_HALF_TO_ODD
_raise_invalid_function_option(round_mode, "round mode")
cdef class _RoundOptions(FunctionOptions):
def _set_options(self, ndigits, round_mode):
self.wrapped.reset(
new CRoundOptions(ndigits, unwrap_round_mode(round_mode))
)
class RoundOptions(_RoundOptions):
def __init__(self, ndigits=0, round_mode="half_to_even"):
self._set_options(ndigits, round_mode)
cdef class _RoundToMultipleOptions(FunctionOptions):
def _set_options(self, multiple, round_mode):
self.wrapped.reset(
new CRoundToMultipleOptions(multiple,
unwrap_round_mode(round_mode))
)
class RoundToMultipleOptions(_RoundToMultipleOptions):
def __init__(self, multiple=1.0, round_mode="half_to_even"):
self._set_options(multiple, round_mode)
cdef class _JoinOptions(FunctionOptions):
_null_handling_map = {
"emit_null": CJoinNullHandlingBehavior_EMIT_NULL,
"skip": CJoinNullHandlingBehavior_SKIP,
"replace": CJoinNullHandlingBehavior_REPLACE,
}
def _set_options(self, null_handling, null_replacement):
try:
self.wrapped.reset(
new CJoinOptions(self._null_handling_map[null_handling],
tobytes(null_replacement))
)
except KeyError:
_raise_invalid_function_option(null_handling, "null handling")
class JoinOptions(_JoinOptions):
def __init__(self, null_handling="emit_null", null_replacement=""):
self._set_options(null_handling, null_replacement)
cdef class _MatchSubstringOptions(FunctionOptions):
def _set_options(self, pattern, ignore_case):
self.wrapped.reset(
new CMatchSubstringOptions(tobytes(pattern), ignore_case)
)
class MatchSubstringOptions(_MatchSubstringOptions):
def __init__(self, pattern, *, ignore_case=False):
self._set_options(pattern, ignore_case)
cdef class _PadOptions(FunctionOptions):
def _set_options(self, width, padding):
self.wrapped.reset(new CPadOptions(width, tobytes(padding)))
class PadOptions(_PadOptions):
def __init__(self, width, padding=' '):
self._set_options(width, padding)
cdef class _TrimOptions(FunctionOptions):
def _set_options(self, characters):
self.wrapped.reset(new CTrimOptions(tobytes(characters)))
class TrimOptions(_TrimOptions):
def __init__(self, characters):
self._set_options(tobytes(characters))
cdef class _ReplaceSliceOptions(FunctionOptions):
def _set_options(self, start, stop, replacement):
self.wrapped.reset(
new CReplaceSliceOptions(start, stop, tobytes(replacement))
)
class ReplaceSliceOptions(_ReplaceSliceOptions):
def __init__(self, start, stop, replacement):
self._set_options(start, stop, replacement)
cdef class _ReplaceSubstringOptions(FunctionOptions):
def _set_options(self, pattern, replacement, max_replacements):
self.wrapped.reset(
new CReplaceSubstringOptions(tobytes(pattern),
tobytes(replacement),
max_replacements)
)
class ReplaceSubstringOptions(_ReplaceSubstringOptions):
def __init__(self, pattern, replacement, *, max_replacements=-1):
self._set_options(pattern, replacement, max_replacements)
cdef class _ExtractRegexOptions(FunctionOptions):
def _set_options(self, pattern):
self.wrapped.reset(new CExtractRegexOptions(tobytes(pattern)))
class ExtractRegexOptions(_ExtractRegexOptions):
def __init__(self, pattern):
self._set_options(pattern)
cdef class _SliceOptions(FunctionOptions):
def _set_options(self, start, stop, step):
self.wrapped.reset(new CSliceOptions(start, stop, step))
class SliceOptions(_SliceOptions):
def __init__(self, start, stop=sys.maxsize, step=1):
self._set_options(start, stop, step)
cdef class _FilterOptions(FunctionOptions):
_null_selection_map = {
"drop": CFilterNullSelectionBehavior_DROP,
"emit_null": CFilterNullSelectionBehavior_EMIT_NULL,
}
def _set_options(self, null_selection_behavior):
try:
self.wrapped.reset(
new CFilterOptions(
self._null_selection_map[null_selection_behavior]
)
)
except KeyError:
_raise_invalid_function_option(null_selection_behavior,
"null selection behavior")
class FilterOptions(_FilterOptions):
def __init__(self, null_selection_behavior="drop"):
self._set_options(null_selection_behavior)
cdef class _DictionaryEncodeOptions(FunctionOptions):
_null_encoding_map = {
"encode": CDictionaryEncodeNullEncodingBehavior_ENCODE,
"mask": CDictionaryEncodeNullEncodingBehavior_MASK,
}
def _set_options(self, null_encoding):
try:
self.wrapped.reset(
new CDictionaryEncodeOptions(
self._null_encoding_map[null_encoding]
)
)
except KeyError:
_raise_invalid_function_option(null_encoding, "null encoding")
class DictionaryEncodeOptions(_DictionaryEncodeOptions):
def __init__(self, null_encoding="mask"):
self._set_options(null_encoding)
cdef class _TakeOptions(FunctionOptions):
def _set_options(self, boundscheck):
self.wrapped.reset(new CTakeOptions(boundscheck))
class TakeOptions(_TakeOptions):
def __init__(self, *, boundscheck=True):
self._set_options(boundscheck)
cdef class _MakeStructOptions(FunctionOptions):
def _set_options(self, field_names, field_nullability, field_metadata):
cdef:
vector[c_string] c_field_names
vector[shared_ptr[const CKeyValueMetadata]] c_field_metadata
for name in field_names:
c_field_names.push_back(tobytes(name))
for metadata in field_metadata:
c_field_metadata.push_back(pyarrow_unwrap_metadata(metadata))
self.wrapped.reset(
new CMakeStructOptions(c_field_names, field_nullability,
c_field_metadata)
)
class MakeStructOptions(_MakeStructOptions):
def __init__(self, field_names, *, field_nullability=None,
field_metadata=None):
if field_nullability is None:
field_nullability = [True] * len(field_names)
if field_metadata is None:
field_metadata = [None] * len(field_names)
self._set_options(field_names, field_nullability, field_metadata)
cdef class _ScalarAggregateOptions(FunctionOptions):
def _set_options(self, skip_nulls, min_count):
self.wrapped.reset(new CScalarAggregateOptions(skip_nulls, min_count))
class ScalarAggregateOptions(_ScalarAggregateOptions):
def __init__(self, *, skip_nulls=True, min_count=1):
self._set_options(skip_nulls, min_count)
cdef class _CountOptions(FunctionOptions):
_mode_map = {
"only_valid": CCountMode_ONLY_VALID,
"only_null": CCountMode_ONLY_NULL,
"all": CCountMode_ALL,
}
def _set_options(self, mode):
try:
self.wrapped.reset(new CCountOptions(self._mode_map[mode]))
except KeyError:
_raise_invalid_function_option(mode, "count mode")
class CountOptions(_CountOptions):
def __init__(self, mode="only_valid"):
self._set_options(mode)
cdef class _IndexOptions(FunctionOptions):
def _set_options(self, scalar):
self.wrapped.reset(new CIndexOptions(pyarrow_unwrap_scalar(scalar)))
class IndexOptions(_IndexOptions):
"""
Options for the index kernel.
Parameters
----------
value : Scalar
The value to search for.
"""
def __init__(self, value):
self._set_options(value)
cdef class _ModeOptions(FunctionOptions):
def _set_options(self, n, skip_nulls, min_count):
self.wrapped.reset(new CModeOptions(n, skip_nulls, min_count))
class ModeOptions(_ModeOptions):
def __init__(self, n=1, *, skip_nulls=True, min_count=0):
self._set_options(n, skip_nulls, min_count)
cdef class _SetLookupOptions(FunctionOptions):
def _set_options(self, value_set, c_bool skip_nulls):
cdef unique_ptr[CDatum] valset
if isinstance(value_set, Array):
valset.reset(new CDatum((<Array> value_set).sp_array))
elif isinstance(value_set, ChunkedArray):
valset.reset(
new CDatum((<ChunkedArray> value_set).sp_chunked_array)
)
elif isinstance(value_set, Scalar):
valset.reset(new CDatum((<Scalar> value_set).unwrap()))
else:
_raise_invalid_function_option(value_set, "value set",
exception_class=TypeError)
self.wrapped.reset(new CSetLookupOptions(deref(valset), skip_nulls))
class SetLookupOptions(_SetLookupOptions):
def __init__(self, value_set, *, skip_nulls=False):
self._set_options(value_set, skip_nulls)
cdef class _StrptimeOptions(FunctionOptions):
_unit_map = {
"s": TimeUnit_SECOND,
"ms": TimeUnit_MILLI,
"us": TimeUnit_MICRO,
"ns": TimeUnit_NANO,
}
def _set_options(self, format, unit):
try:
self.wrapped.reset(
new CStrptimeOptions(tobytes(format), self._unit_map[unit])
)
except KeyError:
_raise_invalid_function_option(unit, "time unit")
class StrptimeOptions(_StrptimeOptions):
def __init__(self, format, unit):
self._set_options(format, unit)
cdef class _StrftimeOptions(FunctionOptions):
def _set_options(self, format, locale):
self.wrapped.reset(
new CStrftimeOptions(tobytes(format), tobytes(locale))
)
class StrftimeOptions(_StrftimeOptions):
def __init__(self, format="%Y-%m-%dT%H:%M:%S", locale="C"):
self._set_options(format, locale)
cdef class _DayOfWeekOptions(FunctionOptions):
def _set_options(self, count_from_zero, week_start):
self.wrapped.reset(
new CDayOfWeekOptions(count_from_zero, week_start)
)
class DayOfWeekOptions(_DayOfWeekOptions):
def __init__(self, *, count_from_zero=True, week_start=1):
self._set_options(count_from_zero, week_start)
cdef class _WeekOptions(FunctionOptions):
def _set_options(self, week_starts_monday, count_from_zero,
first_week_is_fully_in_year):
self.wrapped.reset(
new CWeekOptions(week_starts_monday, count_from_zero,
first_week_is_fully_in_year)
)
class WeekOptions(_WeekOptions):
def __init__(self, *, week_starts_monday=True, count_from_zero=False,
first_week_is_fully_in_year=False):
self._set_options(week_starts_monday,
count_from_zero, first_week_is_fully_in_year)
cdef class _AssumeTimezoneOptions(FunctionOptions):
_ambiguous_map = {
"raise": CAssumeTimezoneAmbiguous_AMBIGUOUS_RAISE,
"earliest": CAssumeTimezoneAmbiguous_AMBIGUOUS_EARLIEST,
"latest": CAssumeTimezoneAmbiguous_AMBIGUOUS_LATEST,
}
_nonexistent_map = {
"raise": CAssumeTimezoneNonexistent_NONEXISTENT_RAISE,
"earliest": CAssumeTimezoneNonexistent_NONEXISTENT_EARLIEST,
"latest": CAssumeTimezoneNonexistent_NONEXISTENT_LATEST,
}
def _set_options(self, timezone, ambiguous, nonexistent):
if ambiguous not in self._ambiguous_map:
_raise_invalid_function_option(ambiguous,
"'ambiguous' timestamp handling")
if nonexistent not in self._nonexistent_map:
_raise_invalid_function_option(nonexistent,
"'nonexistent' timestamp handling")
self.wrapped.reset(
new CAssumeTimezoneOptions(tobytes(timezone),
self._ambiguous_map[ambiguous],
self._nonexistent_map[nonexistent])
)
class AssumeTimezoneOptions(_AssumeTimezoneOptions):
def __init__(self, timezone, *, ambiguous="raise", nonexistent="raise"):
self._set_options(timezone, ambiguous, nonexistent)
cdef class _NullOptions(FunctionOptions):
def _set_options(self, nan_is_null):
self.wrapped.reset(new CNullOptions(nan_is_null))
class NullOptions(_NullOptions):
def __init__(self, *, nan_is_null=False):
self._set_options(nan_is_null)
cdef class _VarianceOptions(FunctionOptions):
def _set_options(self, ddof, skip_nulls, min_count):
self.wrapped.reset(new CVarianceOptions(ddof, skip_nulls, min_count))
class VarianceOptions(_VarianceOptions):
def __init__(self, *, ddof=0, skip_nulls=True, min_count=0):
self._set_options(ddof, skip_nulls, min_count)
cdef class _SplitOptions(FunctionOptions):
def _set_options(self, max_splits, reverse):
self.wrapped.reset(new CSplitOptions(max_splits, reverse))
class SplitOptions(_SplitOptions):
def __init__(self, *, max_splits=-1, reverse=False):
self._set_options(max_splits, reverse)
cdef class _SplitPatternOptions(FunctionOptions):
def _set_options(self, pattern, max_splits, reverse):
self.wrapped.reset(
new CSplitPatternOptions(tobytes(pattern), max_splits, reverse)
)
class SplitPatternOptions(_SplitPatternOptions):
def __init__(self, pattern, *, max_splits=-1, reverse=False):
self._set_options(pattern, max_splits, reverse)
cdef CSortOrder unwrap_sort_order(order) except *:
if order == "ascending":
return CSortOrder_Ascending
elif order == "descending":
return CSortOrder_Descending
_raise_invalid_function_option(order, "sort order")
cdef CNullPlacement unwrap_null_placement(null_placement) except *:
if null_placement == "at_start":
return CNullPlacement_AtStart
elif null_placement == "at_end":
return CNullPlacement_AtEnd
_raise_invalid_function_option(null_placement, "null placement")
cdef class _PartitionNthOptions(FunctionOptions):
def _set_options(self, pivot, null_placement):
self.wrapped.reset(new CPartitionNthOptions(
pivot, unwrap_null_placement(null_placement)))
class PartitionNthOptions(_PartitionNthOptions):
def __init__(self, pivot, *, null_placement="at_end"):
self._set_options(pivot, null_placement)
cdef class _ArraySortOptions(FunctionOptions):
def _set_options(self, order, null_placement):
self.wrapped.reset(new CArraySortOptions(
unwrap_sort_order(order), unwrap_null_placement(null_placement)))
class ArraySortOptions(_ArraySortOptions):
def __init__(self, order="ascending", *, null_placement="at_end"):
self._set_options(order, null_placement)
cdef class _SortOptions(FunctionOptions):
def _set_options(self, sort_keys, null_placement):
cdef vector[CSortKey] c_sort_keys
for name, order in sort_keys:
c_sort_keys.push_back(
CSortKey(tobytes(name), unwrap_sort_order(order))
)
self.wrapped.reset(new CSortOptions(
c_sort_keys, unwrap_null_placement(null_placement)))
class SortOptions(_SortOptions):
def __init__(self, sort_keys, *, null_placement="at_end"):
self._set_options(sort_keys, null_placement)
cdef class _SelectKOptions(FunctionOptions):
def _set_options(self, k, sort_keys):
cdef vector[CSortKey] c_sort_keys
for name, order in sort_keys:
c_sort_keys.push_back(
CSortKey(tobytes(name), unwrap_sort_order(order))
)
self.wrapped.reset(new CSelectKOptions(k, c_sort_keys))
class SelectKOptions(_SelectKOptions):
def __init__(self, k, sort_keys):
self._set_options(k, sort_keys)
cdef class _QuantileOptions(FunctionOptions):
_interp_map = {
"linear": CQuantileInterp_LINEAR,
"lower": CQuantileInterp_LOWER,
"higher": CQuantileInterp_HIGHER,
"nearest": CQuantileInterp_NEAREST,
"midpoint": CQuantileInterp_MIDPOINT,
}
def _set_options(self, quantiles, interp, skip_nulls, min_count):
try:
self.wrapped.reset(
new CQuantileOptions(quantiles, self._interp_map[interp],
skip_nulls, min_count)
)
except KeyError:
_raise_invalid_function_option(interp, "quantile interpolation")
class QuantileOptions(_QuantileOptions):
def __init__(self, q=0.5, *, interpolation="linear", skip_nulls=True,
min_count=0):
if not isinstance(q, (list, tuple, np.ndarray)):
q = [q]
self._set_options(q, interpolation, skip_nulls, min_count)
cdef class _TDigestOptions(FunctionOptions):
def _set_options(self, quantiles, delta, buffer_size, skip_nulls,
min_count):
self.wrapped.reset(
new CTDigestOptions(quantiles, delta, buffer_size, skip_nulls,
min_count)
)
class TDigestOptions(_TDigestOptions):
def __init__(self, q=0.5, *, delta=100, buffer_size=500, skip_nulls=True,
min_count=0):
if not isinstance(q, (list, tuple, np.ndarray)):
q = [q]
self._set_options(q, delta, buffer_size, skip_nulls, min_count)
|