Skip to content

Commit a4eddd1

Browse files
lxy-9602lxy-9602
authored
fix: respect parent validity in Arrow nullability checks (#340)
1 parent 292e6e0 commit a4eddd1

3 files changed

Lines changed: 334 additions & 43 deletions

File tree

‎src/paimon/common/utils/arrow/arrow_utils.cpp‎

Lines changed: 150 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,6 @@
3939
#include "arrow/util/compression.h"
4040
#include "fmt/format.h"
4141
#include "paimon/common/utils/arrow/status_utils.h"
42-
#include "paimon/common/utils/arrow/vector_utils.h"
4342
#include "paimon/common/utils/checked_cast.h"
4443
#include "paimon/common/utils/string_utils.h"
4544
#include "paimon/core/casting/casting_utils.h"
@@ -97,6 +96,153 @@ bool HasUndeclaredDictionaryChild(const std::shared_ptr<arrow::DataType>& type,
9796
return false;
9897
}
9998

99+
// Positions in the current array which remain visible through every nullable ancestor.
100+
struct VisibleRange {
101+
int64_t offset;
102+
int64_t length;
103+
};
104+
105+
using VisibleRanges = std::vector<VisibleRange>;
106+
107+
void AppendVisibleRange(int64_t offset, int64_t length, VisibleRanges* ranges) {
108+
if (length == 0) {
109+
return;
110+
}
111+
if (!ranges->empty()) {
112+
VisibleRange& last = ranges->back();
113+
if (last.offset + last.length == offset) {
114+
last.length += length;
115+
return;
116+
}
117+
}
118+
ranges->push_back({offset, length});
119+
}
120+
121+
Status NullabilityMismatch(const arrow::Field& field) {
122+
return Status::Invalid(fmt::format(
123+
"CheckNullabilityMatch failed, field {} not nullable while data have null value",
124+
field.name()));
125+
}
126+
127+
Result<VisibleRanges> IntersectWithValidity(const arrow::Array& array,
128+
const VisibleRanges& visible_ranges,
129+
const arrow::Field& field) {
130+
const bool nullable = field.nullable();
131+
if (visible_ranges.empty()) {
132+
return VisibleRanges{};
133+
}
134+
135+
const int64_t null_count = array.null_count();
136+
if (null_count == 0) {
137+
return visible_ranges;
138+
}
139+
if (null_count == array.length()) {
140+
if (!nullable) {
141+
return NullabilityMismatch(field);
142+
}
143+
return VisibleRanges{};
144+
}
145+
146+
VisibleRanges valid_ranges;
147+
for (const VisibleRange& range : visible_ranges) {
148+
int64_t run_start = -1;
149+
const int64_t range_end = range.offset + range.length;
150+
for (int64_t i = range.offset; i < range_end; ++i) {
151+
if (array.IsValid(i)) {
152+
if (run_start == -1) {
153+
run_start = i;
154+
}
155+
} else {
156+
if (!nullable) {
157+
return NullabilityMismatch(field);
158+
}
159+
if (run_start != -1) {
160+
AppendVisibleRange(run_start, i - run_start, &valid_ranges);
161+
run_start = -1;
162+
}
163+
}
164+
}
165+
if (run_start != -1) {
166+
AppendVisibleRange(run_start, range_end - run_start, &valid_ranges);
167+
}
168+
}
169+
return valid_ranges;
170+
}
171+
172+
template <typename ListArray>
173+
VisibleRanges GetVisibleValueRanges(const ListArray& array,
174+
const VisibleRanges& visible_parent_ranges) {
175+
VisibleRanges value_ranges;
176+
for (const VisibleRange& range : visible_parent_ranges) {
177+
int64_t value_offset = array.value_offset(range.offset);
178+
int64_t value_end = array.value_offset(range.offset + range.length);
179+
AppendVisibleRange(value_offset, value_end - value_offset, &value_ranges);
180+
}
181+
return value_ranges;
182+
}
183+
184+
Status CheckFieldNullability(const std::shared_ptr<arrow::Field>& field,
185+
const std::shared_ptr<arrow::Array>& data,
186+
const VisibleRanges& visible_ranges) {
187+
const std::shared_ptr<arrow::DataType>& type = field->type();
188+
const bool needs_child_validation =
189+
type->id() == arrow::Type::STRUCT || type->id() == arrow::Type::LIST ||
190+
type->id() == arrow::Type::FIXED_SIZE_LIST || type->id() == arrow::Type::MAP;
191+
if (!needs_child_validation &&
192+
(field->nullable() || visible_ranges.empty() || data->null_count() == 0)) {
193+
return Status::OK();
194+
}
195+
196+
PAIMON_ASSIGN_OR_RAISE(VisibleRanges valid_ranges,
197+
IntersectWithValidity(*data, visible_ranges, *field));
198+
199+
if (type->id() == arrow::Type::STRUCT) {
200+
auto struct_type = checked_pointer_cast<arrow::StructType>(type);
201+
auto struct_array = checked_pointer_cast<arrow::StructArray>(data);
202+
for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
203+
PAIMON_RETURN_NOT_OK(
204+
CheckFieldNullability(struct_type->field(i), struct_array->field(i), valid_ranges));
205+
}
206+
} else if (type->id() == arrow::Type::LIST) {
207+
auto list_type = checked_pointer_cast<arrow::ListType>(type);
208+
auto list_array = checked_pointer_cast<arrow::ListArray>(data);
209+
VisibleRanges value_ranges = GetVisibleValueRanges(*list_array, valid_ranges);
210+
PAIMON_RETURN_NOT_OK(
211+
CheckFieldNullability(list_type->value_field(), list_array->values(), value_ranges));
212+
} else if (type->id() == arrow::Type::FIXED_SIZE_LIST) {
213+
auto vector_type = checked_pointer_cast<arrow::FixedSizeListType>(type);
214+
auto vector_array = checked_pointer_cast<arrow::FixedSizeListArray>(data);
215+
int32_t vector_length = vector_type->list_size();
216+
int64_t required_values = (vector_array->offset() + vector_array->length()) * vector_length;
217+
if (vector_array->values()->length() < required_values) {
218+
return Status::Invalid(fmt::format(
219+
"VECTOR field {} is invalid: VECTOR holds {} elements while {} rows of dimension "
220+
"{} require {}",
221+
field->name(), vector_array->values()->length(), vector_array->length(),
222+
vector_length, required_values));
223+
}
224+
225+
VisibleRanges value_ranges = GetVisibleValueRanges(*vector_array, valid_ranges);
226+
// Paimon VECTOR values cannot contain null elements, irrespective of the Arrow child
227+
// field's declared nullability.
228+
std::shared_ptr<arrow::Field> value_field = vector_type->value_field()->WithNullable(false);
229+
Status status = CheckFieldNullability(value_field, vector_array->values(), value_ranges);
230+
if (!status.ok()) {
231+
return Status::Invalid(
232+
fmt::format("VECTOR field {} is invalid: {}", field->name(), status.message()));
233+
}
234+
} else if (type->id() == arrow::Type::MAP) {
235+
auto map_type = checked_pointer_cast<arrow::MapType>(type);
236+
auto map_array = checked_pointer_cast<arrow::MapArray>(data);
237+
VisibleRanges value_ranges = GetVisibleValueRanges(*map_array, valid_ranges);
238+
PAIMON_RETURN_NOT_OK(
239+
CheckFieldNullability(map_type->key_field(), map_array->keys(), value_ranges));
240+
PAIMON_RETURN_NOT_OK(
241+
CheckFieldNullability(map_type->item_field(), map_array->items(), value_ranges));
242+
}
243+
return Status::OK();
244+
}
245+
100246
bool NeedsNormalization(const std::shared_ptr<arrow::ArrayData>& data) {
101247
if (data->offset != 0) {
102248
return true;
@@ -407,7 +553,9 @@ Status ArrowUtils::CheckNullabilityMatch(const std::shared_ptr<arrow::Schema>& s
407553
struct_array->num_fields(), schema->num_fields()));
408554
}
409555
for (int32_t i = 0; i < schema->num_fields(); i++) {
410-
PAIMON_RETURN_NOT_OK(InnerCheckNullabilityMatch(schema->field(i), struct_array->field(i)));
556+
const std::shared_ptr<arrow::Array>& field_array = struct_array->field(i);
557+
PAIMON_RETURN_NOT_OK(
558+
CheckFieldNullability(schema->field(i), field_array, {{0, field_array->length()}}));
411559
}
412560
return Status::OK();
413561
}
@@ -533,42 +681,6 @@ bool ArrowUtils::EqualsIgnoreNullable(const std::shared_ptr<arrow::DataType>& ty
533681
return true;
534682
}
535683

536-
Status ArrowUtils::InnerCheckNullabilityMatch(const std::shared_ptr<arrow::Field>& field,
537-
const std::shared_ptr<arrow::Array>& data) {
538-
if (PAIMON_UNLIKELY(!field->nullable() && data->null_count() != 0)) {
539-
return Status::Invalid(fmt::format(
540-
"CheckNullabilityMatch failed, field {} not nullable while data have null value",
541-
field->name()));
542-
}
543-
auto type = field->type();
544-
if (type->id() == arrow::Type::STRUCT) {
545-
auto struct_type = checked_pointer_cast<arrow::StructType>(field->type());
546-
auto struct_array = checked_pointer_cast<arrow::StructArray>(data);
547-
for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
548-
PAIMON_RETURN_NOT_OK(
549-
InnerCheckNullabilityMatch(struct_type->field(i), struct_array->field(i)));
550-
}
551-
} else if (type->id() == arrow::Type::LIST) {
552-
auto list_type = checked_pointer_cast<arrow::ListType>(field->type());
553-
auto list_array = checked_pointer_cast<arrow::ListArray>(data);
554-
PAIMON_RETURN_NOT_OK(
555-
InnerCheckNullabilityMatch(list_type->value_field(), list_array->values()));
556-
} else if (type->id() == arrow::Type::FIXED_SIZE_LIST) {
557-
Status status = VectorUtils::ValidateVectorElements(*data);
558-
if (!status.ok()) {
559-
return Status::Invalid(
560-
fmt::format("VECTOR field {} is invalid: {}", field->name(), status.message()));
561-
}
562-
} else if (type->id() == arrow::Type::MAP) {
563-
auto map_type = checked_pointer_cast<arrow::MapType>(field->type());
564-
auto map_array = checked_pointer_cast<arrow::MapArray>(data);
565-
PAIMON_RETURN_NOT_OK(InnerCheckNullabilityMatch(map_type->key_field(), map_array->keys()));
566-
PAIMON_RETURN_NOT_OK(
567-
InnerCheckNullabilityMatch(map_type->item_field(), map_array->items()));
568-
}
569-
return Status::OK();
570-
}
571-
572684
Result<std::shared_ptr<arrow::StructArray>> ArrowUtils::RemoveFieldFromStructArray(
573685
const std::shared_ptr<arrow::StructArray>& struct_array, const std::string& field_name) {
574686
auto struct_type = checked_pointer_cast<arrow::StructType>(struct_array->type());

‎src/paimon/common/utils/arrow/arrow_utils.h‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,8 @@ class PAIMON_EXPORT ArrowUtils {
4343
static Result<std::vector<int32_t>> CreateProjection(
4444
const std::shared_ptr<arrow::Schema>& src_schema, const arrow::FieldVector& target_fields);
4545

46+
/// Checks that every logically visible null matches the schema. Values hidden by a null
47+
/// STRUCT, LIST, MAP, or FIXED_SIZE_LIST parent are ignored.
4648
static Status CheckNullabilityMatch(const std::shared_ptr<arrow::Schema>& schema,
4749
const std::shared_ptr<arrow::Array>& data);
4850

@@ -149,10 +151,6 @@ class PAIMON_EXPORT ArrowUtils {
149151
const std::shared_ptr<arrow::StructArray>& batch,
150152
const std::shared_ptr<arrow::DataType>& logical_type, arrow::MemoryPool* pool,
151153
bool preserve_layout_recoverable_dictionaries);
152-
153-
private:
154-
static Status InnerCheckNullabilityMatch(const std::shared_ptr<arrow::Field>& field,
155-
const std::shared_ptr<arrow::Array>& data);
156154
};
157155

158156
} // namespace paimon

0 commit comments

Comments
 (0)