fix filter processing checker (#16476)

ivanmorozov333 · web-flow · commit a701bdbc0909 · 2025-03-31T18:25:33.000+03:00
diff --git a/ydb/core/formats/arrow/accessor/abstract/accessor.cpp b/ydb/core/formats/arrow/accessor/abstract/accessor.cpp
@@ -62,10 +62,10 @@ IChunkedArray::TFullDataAddress IChunkedArray::GetChunk(const std::optional<TAdd
         return TFullDataAddress(localAddress.GetArray(), std::move(addressChain));
     } else {
         auto chunkedArrayAddress = GetArray(chunkCurrent, position, nullptr);
-        if (chunkCurrent) {
-            AFL_VERIFY(chunkCurrent->GetSize() == 1 + chunkedArrayAddress.GetAddress().GetSize())("current", chunkCurrent->GetSize())(
-                                                      "chunked", chunkedArrayAddress.GetAddress().GetSize());
-        }
+//        if (chunkCurrent) {
+//            AFL_VERIFY(chunkCurrent->GetSize() == chunkedArrayAddress.GetAddress().GetSize())("current", chunkCurrent->GetSize())(
+//                                                      "chunked", chunkedArrayAddress.GetAddress().GetSize());
+//        }
         auto localAddress = chunkedArrayAddress.GetArray()->GetLocalData(address, chunkedArrayAddress.GetAddress().GetLocalIndex(position));
         auto fullAddress = std::move(chunkedArrayAddress.MutableAddress());
         fullAddress.Add(localAddress.GetAddress());
@@ -112,7 +112,7 @@ std::shared_ptr<IChunkedArray> IChunkedArray::DoApplyFilter(const TColumnFilter&
     auto schema = std::make_shared<arrow::Schema>(fields);
     auto table = arrow::Table::Make(schema, { arr }, GetRecordsCount());
     AFL_VERIFY(table->num_columns() == 1);
-    AFL_VERIFY(filter.Apply(table));
+    filter.Apply(table);
     if (table->column(0)->num_chunks() == 1) {
         return std::make_shared<TTrivialArray>(table->column(0)->chunk(0));
     } else {
diff --git a/ydb/core/formats/arrow/accessor/sub_columns/columns_storage.cpp b/ydb/core/formats/arrow/accessor/sub_columns/columns_storage.cpp
@@ -31,7 +31,7 @@ TColumnsData TColumnsData::ApplyFilter(const TColumnFilter& filter) const {
         return *this;
     }
     auto records = Records;
-    AFL_VERIFY(filter.Apply(records));
+    filter.Apply(records);
     if (records->GetRecordsCount()) {
         TDictStats::TBuilder builder;
         ui32 idx = 0;
diff --git a/ydb/core/formats/arrow/accessor/sub_columns/others_storage.cpp b/ydb/core/formats/arrow/accessor/sub_columns/others_storage.cpp
@@ -253,7 +253,7 @@ std::shared_ptr<IChunkedArray> TOthersData::GetPathAccessor(const std::string_vi
         filter.Add(it.GetKeyIndex() == *idx);
     }
     auto recordsFiltered = Records;
-    AFL_VERIFY(filter.Apply(recordsFiltered));
+    filter.Apply(recordsFiltered);
     auto table = recordsFiltered->BuildTableVerified(std::set<std::string>({ "record_idx", "value" }));
 
     TSparsedArray::TBuilder builder(nullptr, arrow::utf8());
diff --git a/ydb/core/formats/arrow/arrow_filter.cpp b/ydb/core/formats/arrow/arrow_filter.cpp
@@ -363,9 +363,9 @@ NKikimr::NArrow::TColumnFilter TColumnFilter::MakePredicateFilter(
 }
 
 template <class TData>
-bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const TColumnFilter::TApplyContext& context) {
+void ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const TColumnFilter::TApplyContext& context) {
     if (!batch || !batch->num_rows()) {
-        return false;
+        return;
     }
     if (!filter.IsEmpty()) {
         if (context.HasSlice()) {
@@ -380,10 +380,10 @@ bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const
     }
     if (filter.IsTotalDenyFilter()) {
         batch = NAdapter::TDataBuilderPolicy<TData>::GetEmptySame(batch);
-        return true;
+        return;
     }
     if (filter.IsTotalAllowFilter()) {
-        return true;
+        return;
     }
     if (context.GetTrySlices() && filter.GetFilter().size() * 10 < filter.GetRecordsCountVerified() &&
         filter.GetRecordsCountVerified() < filter.GetFilteredCountVerified() * 50) {
@@ -394,18 +394,17 @@ bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const
     } else {
         batch = NAdapter::TDataBuilderPolicy<TData>::ApplyArrowFilter(batch, filter);
     }
-    return batch->num_rows();
 }
 
-bool TColumnFilter::Apply(std::shared_ptr<TGeneralContainer>& batch, const TApplyContext& context) const {
+void TColumnFilter::Apply(std::shared_ptr<TGeneralContainer>& batch, const TApplyContext& context) const {
     return ApplyImpl(*this, batch, context);
 }
 
-bool TColumnFilter::Apply(std::shared_ptr<arrow::Table>& batch, const TApplyContext& context) const {
+void TColumnFilter::Apply(std::shared_ptr<arrow::Table>& batch, const TApplyContext& context) const {
     return ApplyImpl(*this, batch, context);
 }
 
-bool TColumnFilter::Apply(std::shared_ptr<arrow::RecordBatch>& batch, const TApplyContext& context) const {
+void TColumnFilter::Apply(std::shared_ptr<arrow::RecordBatch>& batch, const TApplyContext& context) const {
     return ApplyImpl(*this, batch, context);
 }
 
diff --git a/ydb/core/formats/arrow/arrow_filter.h b/ydb/core/formats/arrow/arrow_filter.h
@@ -289,9 +289,9 @@ class TColumnFilter {
         TApplyContext& Slice(const ui32 start, const ui32 count);
     };
 
-    [[nodiscard]] bool Apply(std::shared_ptr<TGeneralContainer>& batch, const TApplyContext& context = Default<TApplyContext>()) const;
-    [[nodiscard]] bool Apply(std::shared_ptr<arrow::Table>& batch, const TApplyContext& context = Default<TApplyContext>()) const;
-    [[nodiscard]] bool Apply(std::shared_ptr<arrow::RecordBatch>& batch, const TApplyContext& context = Default<TApplyContext>()) const;
+    void Apply(std::shared_ptr<TGeneralContainer>& batch, const TApplyContext& context = Default<TApplyContext>()) const;
+    void Apply(std::shared_ptr<arrow::Table>& batch, const TApplyContext& context = Default<TApplyContext>()) const;
+    void Apply(std::shared_ptr<arrow::RecordBatch>& batch, const TApplyContext& context = Default<TApplyContext>()) const;
     void Apply(const ui32 expectedRecordsCount, std::vector<arrow::Datum*>& datums) const;
     [[nodiscard]] std::shared_ptr<NAccessor::IChunkedArray> Apply(
         const std::shared_ptr<NAccessor::IChunkedArray>& source, const TApplyContext& context = Default<TApplyContext>()) const;
diff --git a/ydb/core/formats/arrow/reader/merger.cpp b/ydb/core/formats/arrow/reader/merger.cpp
@@ -105,7 +105,7 @@ std::shared_ptr<arrow::Table> TMergePartialStream::SingleSourceDrain(const TSort
             *lastResultPosition = TCursor(keys, 0, SortSchema->field_names());
         }
         if (SortHeap.Current().GetFilter()) {
-            AFL_VERIFY(SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(pos.GetPosition() + (include ? 0 : 1), resultSize)));
+            SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(pos.GetPosition() + (include ? 0 : 1), resultSize));
         }
     } else {
         result = SortHeap.Current().GetKeyColumns().SliceData(startPos, resultSize);
@@ -114,7 +114,7 @@ std::shared_ptr<arrow::Table> TMergePartialStream::SingleSourceDrain(const TSort
             *lastResultPosition = TCursor(keys, keys->num_rows() - 1, SortSchema->field_names());
         }
         if (SortHeap.Current().GetFilter()) {
-            AFL_VERIFY(SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(startPos, resultSize)));
+            SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(startPos, resultSize));
         }
     }
     if (!result || !result->num_rows()) {
diff --git a/ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/fetching.cpp b/ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/fetching.cpp
@@ -159,7 +159,10 @@ TConclusion<bool> TBuildResultStep::DoExecuteInplace(const std::shared_ptr<IData
     if (!source->GetStageResult().IsEmpty()) {
         resultBatch = source->GetStageResult().GetBatch()->BuildTableVerified(contextTableConstruct);
         if (auto filter = source->GetStageResult().GetNotAppliedFilter()) {
-            AFL_VERIFY(filter->Apply(resultBatch, NArrow::TColumnFilter::TApplyContext(StartIndex, RecordsCount).SetTrySlices(true)));
+            filter->Apply(resultBatch, NArrow::TColumnFilter::TApplyContext(StartIndex, RecordsCount).SetTrySlices(true));
+            if (!resultBatch->num_rows()) {
+                resultBatch = nullptr;
+            }
         }
     }
 
diff --git a/ydb/core/tx/columnshard/engines/reader/sys_view/abstract/iterator.cpp b/ydb/core/tx/columnshard/engines/reader/sys_view/abstract/iterator.cpp
@@ -45,7 +45,7 @@ TConclusion<std::shared_ptr<TPartialReadResult>> TStatsIteratorBase::GetBatch()
 
         {
             NArrow::TColumnFilter filter = ReadMetadata->GetPKRangesFilter().BuildFilter(originalBatch);
-            AFL_VERIFY(filter.Apply(originalBatch));
+            filter.Apply(originalBatch);
         }
 
         // Leave only requested columns
diff --git a/ydb/core/tx/columnshard/operations/batch_builder/merger.h b/ydb/core/tx/columnshard/operations/batch_builder/merger.h
@@ -67,7 +67,7 @@ class TReplaceMerger: public IMerger {
 
     virtual NArrow::TContainerWithIndexes<arrow::RecordBatch> BuildResultBatch() override {
         auto result = IncomingData;
-        AFL_VERIFY(Filter.Apply(result.MutableContainer()));
+        Filter.Apply(result.MutableContainer());
         return result;
     }
 };

Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,7 @@ TColumnsData TColumnsData::ApplyFilter(const TColumnFilter& filter) const {`
`31`	`31`	`return *this;`
`32`	`32`	`}`
`33`	`33`	`auto records = Records;`
`34`		`- AFL_VERIFY(filter.Apply(records));`
	`34`	`+ filter.Apply(records);`
`35`	`35`	`if (records->GetRecordsCount()) {`
`36`	`36`	`TDictStats::TBuilder builder;`
`37`	`37`	`ui32 idx = 0;`
Original file line number	Diff line number	Diff line change
`@@ -253,7 +253,7 @@ std::shared_ptr<IChunkedArray> TOthersData::GetPathAccessor(const std::string_vi`
`253`	`253`	`filter.Add(it.GetKeyIndex() == *idx);`
`254`	`254`	`}`
`255`	`255`	`auto recordsFiltered = Records;`
`256`		`- AFL_VERIFY(filter.Apply(recordsFiltered));`
	`256`	`+ filter.Apply(recordsFiltered);`
`257`	`257`	`auto table = recordsFiltered->BuildTableVerified(std::set<std::string>({ "record_idx", "value" }));`
`258`	`258`
`259`	`259`	`TSparsedArray::TBuilder builder(nullptr, arrow::utf8());`
Original file line number	Diff line number	Diff line change
`@@ -363,9 +363,9 @@ NKikimr::NArrow::TColumnFilter TColumnFilter::MakePredicateFilter(`
`363`	`363`	`}`
`364`	`364`
`365`	`365`	`template <class TData>`
`366`		`-bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const TColumnFilter::TApplyContext& context) {`
	`366`	`+void ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const TColumnFilter::TApplyContext& context) {`
`367`	`367`	`if (!batch \|\| !batch->num_rows()) {`
`368`		`- return false;`
	`368`	`+ return;`
`369`	`369`	`}`
`370`	`370`	`if (!filter.IsEmpty()) {`
`371`	`371`	`if (context.HasSlice()) {`
`@@ -380,10 +380,10 @@ bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const`
`380`	`380`	`}`
`381`	`381`	`if (filter.IsTotalDenyFilter()) {`
`382`	`382`	`batch = NAdapter::TDataBuilderPolicy<TData>::GetEmptySame(batch);`
`383`		`- return true;`
	`383`	`+ return;`
`384`	`384`	`}`
`385`	`385`	`if (filter.IsTotalAllowFilter()) {`
`386`		`- return true;`
	`386`	`+ return;`
`387`	`387`	`}`
`388`	`388`	`if (context.GetTrySlices() && filter.GetFilter().size() * 10 < filter.GetRecordsCountVerified() &&`
`389`	`389`	`filter.GetRecordsCountVerified() < filter.GetFilteredCountVerified() * 50) {`
`@@ -394,18 +394,17 @@ bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const`
`394`	`394`	`} else {`
`395`	`395`	`batch = NAdapter::TDataBuilderPolicy<TData>::ApplyArrowFilter(batch, filter);`
`396`	`396`	`}`
`397`		`- return batch->num_rows();`
`398`	`397`	`}`
`399`	`398`
`400`		`-bool TColumnFilter::Apply(std::shared_ptr<TGeneralContainer>& batch, const TApplyContext& context) const {`
	`399`	`+void TColumnFilter::Apply(std::shared_ptr<TGeneralContainer>& batch, const TApplyContext& context) const {`
`401`	`400`	`return ApplyImpl(*this, batch, context);`
`402`	`401`	`}`
`403`	`402`
`404`		`-bool TColumnFilter::Apply(std::shared_ptr<arrow::Table>& batch, const TApplyContext& context) const {`
	`403`	`+void TColumnFilter::Apply(std::shared_ptr<arrow::Table>& batch, const TApplyContext& context) const {`
`405`	`404`	`return ApplyImpl(*this, batch, context);`
`406`	`405`	`}`
`407`	`406`
`408`		`-bool TColumnFilter::Apply(std::shared_ptr<arrow::RecordBatch>& batch, const TApplyContext& context) const {`
	`407`	`+void TColumnFilter::Apply(std::shared_ptr<arrow::RecordBatch>& batch, const TApplyContext& context) const {`
`409`	`408`	`return ApplyImpl(*this, batch, context);`
`410`	`409`	`}`
`411`	`410`
Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ std::shared_ptr<arrow::Table> TMergePartialStream::SingleSourceDrain(const TSort`
`105`	`105`	`*lastResultPosition = TCursor(keys, 0, SortSchema->field_names());`
`106`	`106`	`}`
`107`	`107`	`if (SortHeap.Current().GetFilter()) {`
`108`		`- AFL_VERIFY(SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(pos.GetPosition() + (include ? 0 : 1), resultSize)));`
	`108`	`+ SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(pos.GetPosition() + (include ? 0 : 1), resultSize));`
`109`	`109`	`}`
`110`	`110`	`} else {`
`111`	`111`	`result = SortHeap.Current().GetKeyColumns().SliceData(startPos, resultSize);`
`@@ -114,7 +114,7 @@ std::shared_ptr<arrow::Table> TMergePartialStream::SingleSourceDrain(const TSort`
`114`	`114`	`*lastResultPosition = TCursor(keys, keys->num_rows() - 1, SortSchema->field_names());`
`115`	`115`	`}`
`116`	`116`	`if (SortHeap.Current().GetFilter()) {`
`117`		`- AFL_VERIFY(SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(startPos, resultSize)));`
	`117`	`+ SortHeap.Current().GetFilter()->Apply(result, TColumnFilter::TApplyContext(startPos, resultSize));`
`118`	`118`	`}`
`119`	`119`	`}`
`120`	`120`	`if (!result \|\| !result->num_rows()) {`
Original file line number	Diff line number	Diff line change
`@@ -159,7 +159,10 @@ TConclusion<bool> TBuildResultStep::DoExecuteInplace(const std::shared_ptr<IData`
`159`	`159`	`if (!source->GetStageResult().IsEmpty()) {`
`160`	`160`	`resultBatch = source->GetStageResult().GetBatch()->BuildTableVerified(contextTableConstruct);`
`161`	`161`	`if (auto filter = source->GetStageResult().GetNotAppliedFilter()) {`
`162`		`- AFL_VERIFY(filter->Apply(resultBatch, NArrow::TColumnFilter::TApplyContext(StartIndex, RecordsCount).SetTrySlices(true)));`
	`162`	`+ filter->Apply(resultBatch, NArrow::TColumnFilter::TApplyContext(StartIndex, RecordsCount).SetTrySlices(true));`
	`163`	`+ if (!resultBatch->num_rows()) {`
	`164`	`+ resultBatch = nullptr;`
	`165`	`+ }`
`163`	`166`	`}`
`164`	`167`	`}`
`165`	`168`
Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,7 @@ TConclusion<std::shared_ptr<TPartialReadResult>> TStatsIteratorBase::GetBatch()`
`45`	`45`
`46`	`46`	`{`
`47`	`47`	`NArrow::TColumnFilter filter = ReadMetadata->GetPKRangesFilter().BuildFilter(originalBatch);`
`48`		`- AFL_VERIFY(filter.Apply(originalBatch));`
	`48`	`+ filter.Apply(originalBatch);`
`49`	`49`	`}`
`50`	`50`
`51`	`51`	`// Leave only requested columns`
Original file line number	Diff line number	Diff line change
`@@ -67,7 +67,7 @@ class TReplaceMerger: public IMerger {`
`67`	`67`
`68`	`68`	`virtual NArrow::TContainerWithIndexes<arrow::RecordBatch> BuildResultBatch() override {`
`69`	`69`	`auto result = IncomingData;`
`70`		`- AFL_VERIFY(Filter.Apply(result.MutableContainer()));`
	`70`	`+ Filter.Apply(result.MutableContainer());`
`71`	`71`	`return result;`
`72`	`72`	`}`
`73`	`73`	`};`