fix filter apply and serializer usage (ydb-platform#1629)

* fix filter apply

* fixes

Author: ivanmorozov333

ydb/core/formats/arrow/serializer/abstract.cpp

@@ -6,7 +6,7 @@ NKikimr::TConclusionStatus TSerializerContainer::DeserializeFromProto(const NKik
     NKikimrSchemeOp::TOlapColumn::TSerializer serializerProto;
     serializerProto.SetClassName(NArrow::NSerialization::TNativeSerializer::GetClassNameStatic());
     *serializerProto.MutableArrowCompression() = proto;
-    AFL_VERIFY(Initialize(NArrow::NSerialization::TNativeSerializer::GetClassNameStatic()));
+    AFL_VERIFY(Initialize(NArrow::NSerialization::TNativeSerializer::GetClassNameStatic(), true));
     return GetObjectPtr()->DeserializeFromProto(serializerProto);
 }
 

ydb/core/tx/columnshard/engines/reader/plain_reader/fetched_data.h

@@ -63,14 +63,10 @@ class TFetchedData {
     }
 
     void AddFilter(const std::shared_ptr<NArrow::TColumnFilter>& filter) {
-        if (UseFilter && Table && filter) {
-            AFL_VERIFY(filter->Apply(Table));
-        }
-        if (!Filter) {
-            Filter = filter;
-        } else if (filter) {
-            *Filter = Filter->CombineSequentialAnd(*filter);
+        if (!filter) {
+            return;
         }
+        return AddFilter(*filter);
     }
 
     void AddFilter(const NArrow::TColumnFilter& filter) {
@@ -79,8 +75,10 @@ class TFetchedData {
         }
         if (!Filter) {
             Filter = std::make_shared<NArrow::TColumnFilter>(filter);
-        } else {
+        } else if (UseFilter) {
             *Filter = Filter->CombineSequentialAnd(filter);
+        } else {
+            *Filter = Filter->And(filter);
         }
     }
 

ydb/core/tx/columnshard/engines/scheme/column_features.cpp

@@ -33,8 +33,6 @@ std::optional<NKikimr::NOlap::TColumnFeatures> TColumnFeatures::BuildFromProto(c
         AFL_VERIFY(result.Serializer.DeserializeFromProto(columnInfo.GetSerializer()));
     } else if (columnInfo.HasCompression()) {
         AFL_VERIFY(result.Serializer.DeserializeFromProto(columnInfo.GetCompression()));
-    } else {
-        result.Serializer = NArrow::NSerialization::TSerializerContainer::GetDefaultSerializer();
     }
     if (columnInfo.HasDictionaryEncoding()) {
         auto settings = NArrow::NDictionary::TEncodingSettings::BuildFromProto(columnInfo.GetDictionaryEncoding());

ydb/services/bg_tasks/abstract/interface.h

@@ -117,8 +117,8 @@ class TCommonInterfaceContainer {
         static_assert(std::is_base_of<IInterface, TDerived>::value);
     }
 
-    bool Initialize(const TString& className) {
-        AFL_VERIFY(!Object)("problem", "initialize for not-empty-object");
+    bool Initialize(const TString& className, const bool maybeExists = false) {
+        AFL_VERIFY(maybeExists || !Object)("problem", "initialize for not-empty-object");
         Object.reset(TFactory::Construct(className));
         if (!Object) {
             ALS_ERROR(NKikimrServices::BG_TASKS) << "incorrect class name: " << className << " for " << typeid(IInterface).name();